diff --git a/.github/workflows/pylint.yml b/.github/workflows/pylint.yml new file mode 100644 index 0000000..5abdb30 --- /dev/null +++ b/.github/workflows/pylint.yml @@ -0,0 +1,23 @@ +name: Pylint + +on: [push] + +jobs: + build: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.10"] + steps: + - uses: actions/checkout@v3 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v3 + with: + python-version: ${{ matrix.python-version }} + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install pylint + - name: Analysing the code with pylint + run: | + pylint $(git ls-files '*.py') diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5e8862c --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +# Nix +result diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index ac2bdc3..555942d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -11,7 +11,7 @@ repos: # hooks: # - id: prettier - repo: https://github.com/ambv/black - rev: "23.3.0" + rev: "23.7.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.31.0" + rev: "v1.32.0" hooks: - id: yamllint diff --git a/.pylintrc b/.pylintrc new file mode 100644 index 0000000..4eeae7c --- /dev/null +++ b/.pylintrc @@ -0,0 +1,209 @@ +[MASTER] + +# Specify a configuration file. +#rcfile= + +# Python code to execute, usually for sys.path manipulation such as +# pygtk.require(). +#init-hook= + + +# Add to the black list. It should be a base name, not a +# path. You may set this option multiple times. +ignore=.hg + +# Pickle collected data for later comparisons. +persistent=yes + +# List of plugins (as comma separated values of python modules names) to load, +# usually to register additional checkers. +load-plugins= + + +[MESSAGES CONTROL] + +# Enable the message, report, category or checker with the given id(s). You can +# either give multiple identifier separated by comma (,) or put this option +# multiple time. +#enable= + +# Disable the message, report, category or checker with the given id(s). You +# can either give multiple identifier separated by comma (,) or put this option +# multiple time (only on the command line, not in the configuration file where +# it should appear only once). +disable=R0903,W0603 + + +[REPORTS] + +# Set the output format. Available formats are text, parseable, colorized, msvs +# (visual studio) and html +output-format=text +# Tells whether to display a full report or only the messages +reports=yes + +# Python expression which should return a note less than 10 (10 is the highest +# note). You have access to the variables errors warning, statement which +# respectively contain the number of errors / warnings messages and the total +# number of statements analyzed. This is used by the global evaluation report +# (R0004). +evaluation=10.0 - ((float(5 * error + warning + refactor + convention) / statement) * 10) + + + +[BASIC] + + +# Regular expression which should only match correct module names +module-rgx=(([a-z_][a-z0-9_]*)|([A-Z][a-zA-Z0-9]+))$ + +# Regular expression which should only match correct module level names +const-rgx=(([A-Z_][A-Z0-9_]*)|(__.*__)|log(_.*)?)$ + +# Regular expression which should only match correct class names +class-rgx=[A-Z_][a-zA-Z0-9]+$ + +# Regular expression which should only match correct function names +function-rgx=[a-z_][a-zA-Z0-9_]{2,30}$ + +# Regular expression which should only match correct method names +method-rgx=[a-z_][a-zA-Z0-9_]{2,30}$ + +# Regular expression which should only match correct instance attribute names +attr-rgx=[a-z_][a-z0-9_]{2,30}$ + +# Regular expression which should only match correct argument names +argument-rgx=[a-z_][a-z0-9_]{2,30}$ + +# Regular expression which should only match correct variable names +variable-rgx=[a-z_][a-z0-9_]{,30}$ + +# Regular expression which should only match correct list comprehension / +# generator expression variable names +inlinevar-rgx=[A-Za-z_][A-Za-z0-9_]*$ + +# Good variable names which should always be accepted, separated by a comma +good-names=i,j,k,ex,Run,_ + +# Bad variable names which should always be refused, separated by a comma +bad-names=foo,bar,baz,toto,tutu,tata,pdb + +# Regular expression which should only match functions or classes name which do +# not require a docstring +no-docstring-rgx=__.*__ + + +[FORMAT] + +# Maximum number of characters on a single line. +max-line-length=120 + +# Maximum number of lines in a module +max-module-lines=1000 + +# String used as indentation unit. This is usually " " (4 spaces) or "\t" (1 +# tab). +indent-string=' ' + + +[MISCELLANEOUS] + +# List of note tags to take in consideration, separated by a comma. +notes=FIXME,XXX,TODO + + +[SIMILARITIES] + +# Minimum lines number of a similarity. +min-similarity-lines=4 + +# Ignore comments when computing similarities. +ignore-comments=yes + +# Ignore docstrings when computing similarities. +ignore-docstrings=yes + + +[TYPECHECK] + +# Tells whether missing members accessed in mixin class should be ignored. A +# mixin class is detected if its name ends with "mixin" (case insensitive). +ignore-mixin-members=yes + +# List of classes names for which member attributes should not be checked +# (useful for classes with attributes dynamically set). +ignored-classes=SQLObject + + +# List of members which are set dynamically and missed by pylint inference +# system, and so shouldn't trigger E0201 when accessed. +generated-members=REQUEST,acl_users,aq_parent + + +[VARIABLES] + +# Tells whether we should check for unused import in __init__ files. +init-import=no + +# A regular expression matching names used for dummy variables (i.e. not used). +dummy-variables-rgx=_|dummy + +# List of additional names supposed to be defined in builtins. Remember that +# you should avoid to define new builtins when possible. +#additional-builtins= +additional-builtins = _,DBG + + +[CLASSES] + +# List of method names used to declare (i.e. assign) instance attributes. +defining-attr-methods=__init__,__new__,setUp + + +[DESIGN] + +# Maximum number of arguments for function / method +max-args=7 + +# Argument names that match this expression will be ignored. Default to name +# with leading underscore +ignored-argument-names=_.* + +# Maximum number of locals for function / method body +max-locals=15 + +# Maximum number of return / yield for function / method body +max-returns=6 + +# Maximum number of statements in function / method body +max-statements=50 + +# Maximum number of parents for a class (see R0901). +max-parents=7 + +# Maximum number of attributes for a class (see R0902). +max-attributes=10 + +# Minimum number of public methods for a class (see R0903). +min-public-methods=2 + +# Maximum number of public methods for a class (see R0904). +max-public-methods=20 + + +[IMPORTS] + +# Deprecated modules which should not be used, separated by a comma +deprecated-modules=regsub,string,TERMIOS,Bastion,rexec + +# Create a graph of every (i.e. internal and external) dependencies in the +# given file (report RP0402 must not be disabled) +import-graph= + +# Create a graph of external dependencies in the given file (report RP0402 must +# not be disabled) +ext-import-graph= + +# Create a graph of internal dependencies in the given file (report RP0402 must +# not be disabled) +int-import-graph= diff --git a/README.md b/README.md index 460af9a..4b168d9 100644 --- a/README.md +++ b/README.md @@ -1,188 +1,62 @@ -# Extensions & tweaks for hyprland +# Pyprland + +## Scratchpads, smart monitor placement and other 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 - } -} -``` +Check the [wiki](https://github.com/hyprland-community/pyprland/wiki) for more information. -Built-in plugins are: +# 1.4.2 (WIP) -- `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 +- [two new options](https://github.com/hyprland-community/pyprland/wiki/Plugins#size-optional) for scratchpads: `position` and `size` - from @iliayar +- bugfixes -## Installation +# 1.4.1 -``` -pip install pyprland -``` +- minor bugfixes -Don't forget to start the process, for instance: +# 1.4.0 -``` -exec-once = pypr -``` +- 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 +- fix `scratchpads`'s position on monitors using scaling +- improve error handling & logging, enable debug logs with `--debug ` -## Getting started +## 1.3.1 -Create a configuration file in `~/.config/hypr/pyprland.json` enabling a list of plugins, each plugin may have its own configuration needs, eg: +- `monitors` triggers rules on startup (not only when a monitor is plugged) -```json -{ - "pyprland": { - "plugins": [ - "scratchpads", - "monitors", - "workspaces_follow_focus" - ] - }, - "scratchpads": { - "term": { - "command": "kitty --class kitty-dropterm", - "animation": "fromTop", - "unfocus": "hide" - }, - "volume": { - "command": "pavucontrol", - "unfocus": "hide", - "animation": "fromRight" - } - }, - "monitors": { - "placement": { - "BenQ PJ": { - "topOf": "eDP-1" - } - } - } -} -``` +## 1.3.0 -# Configuring plugins +- Add `shift_monitors` addon +- Add `monitors` addon +- scratchpads: more reliable client tracking +- bugfixes -## `monitors` +## 1.2.1 -Requires `wlr-randr`. +- scratchpads have their own special workspaces now +- misc improvements -Allows relative placement of monitors depending on the model ("description" returned by `hyprctl monitors`). +## 1.2.0 -### Configuration +- Add `magnify` addon +- focus fix when closing a scratchpad +- misc improvements -Supported placements are: +## 1.1.0 -- leftOf -- topOf -- rightOf -- bottomOf +- Add `lost_windows` addon +- Add `toggle_dpms` addon +- `workspaces_follow_focus` now requires hyprland 0.25.0 +- misc improvements -## `workspaces_follow_focus` +## 1.0.1, 1.0.2 -Make non-visible workspaces follow the focused monitor. -Also provides commands to switch between workspaces wile preserving the current monitor assignments: +- bugfixes & improvements -### Commands +## 1.0 -- `change_workspace` ``: changes the workspace of the focused monitor +- First release, a modular hpr-scratcher (`scratchpads` plugin) +- Add `workspaces_follow_focus` addon -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 ` : toggle the given scratchpad -- `show ` : show the given scratchpad -- `hide ` : 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 diff --git a/flake.lock b/flake.lock new file mode 100644 index 0000000..6067f6e --- /dev/null +++ b/flake.lock @@ -0,0 +1,61 @@ +{ + "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 +} diff --git a/flake.nix b/flake.nix new file mode 100644 index 0000000..b60c2b8 --- /dev/null +++ b/flake.nix @@ -0,0 +1,27 @@ +{ + 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; + }; + } + ); +} diff --git a/poetry.lock b/poetry.lock new file mode 100644 index 0000000..022bbde --- /dev/null +++ b/poetry.lock @@ -0,0 +1,7 @@ +# 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" diff --git a/pyprland/command.py b/pyprland/command.py index 9fb50d3..17ffd7c 100755 --- a/pyprland/command.py +++ b/pyprland/command.py @@ -1,14 +1,16 @@ #!/bin/env python +""" Pyprland - an Hyprland companion app (cli client & daemon) """ import asyncio -import json -import sys -import os import importlib -import traceback - +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 .common import DEBUG +from .ipc import init as ipc_init from .plugins.interface import Plugin CONTROL = f'/tmp/hypr/{ os.environ["HYPRLAND_INSTANCE_SIGNATURE"] }/.pyprland.sock' @@ -17,58 +19,92 @@ CONFIG_FILE = "~/.config/hypr/pyprland.json" class Pyprland: + "Main app object" server: asyncio.Server event_reader: asyncio.StreamReader stopped = False name = "builtin" + config: None | dict[str, dict] = None def __init__(self): self.plugins: dict[str, Plugin] = {} + self.log = get_logger() - 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"]: + async def load_config(self, init=True): + """Loads the configuration + + if `init` is true, also initializes the plugins""" + try: + with open(os.path.expanduser(CONFIG_FILE), encoding="utf-8") as f: + self.config = json.loads(f.read()) + except FileNotFoundError as e: + self.log.critical( + "No config file found, create one at ~/.config/hypr/pyprland.json with a valid pyprland.plugins list" + ) + raise PyprError() from e + + assert self.config + + for name in cast(dict, 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).Extension(name) - await plug.init() + if init: + 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) + self.log.error("Error loading plugin %s:", name, exc_info=True) + raise PyprError() from e + if init: + try: + await self.plugins[name].load_config(self.config) + except PyprError: + raise + except Exception as e: + self.log.error("Error initializing plugin %s:", name, exc_info=True) + raise PyprError() from e async def _callHandler(self, full_name, *params): + "Call an event handler with params" + for plugin in [self] + list(self.plugins.values()): if hasattr(plugin, full_name): + self.log.debug("%s.%s%s", plugin.name, full_name, params) try: await getattr(plugin, full_name)(*params) - except Exception as e: - print(f"{plugin.name}::{full_name}({params}) failed:") - if DEBUG: - traceback.print_exc() + 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 + self.log.warning( + "%s::%s(%s) failed:", plugin.name, full_name, params + ) + self.log.exception(e) async def read_events_loop(self): + "Consumes the event loop and calls corresponding handlers" while not self.stopped: - data = (await self.event_reader.readline()).decode() + try: + data = (await self.event_reader.readline()).decode() + except UnicodeDecodeError: + self.log.error("Invalid unicode while reading events") + continue if not data: - print("Reader starved") + self.log.critical("Reader starved") return - cmd, params = data.split(">>") + cmd, params = data.split(">>", 1) 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) -> None: + "Receives a socket command" data = (await reader.readline()).decode() if not data: - print("Server starved") + self.log.critical("Server starved") return if data == "exit\n": self.stopped = True @@ -85,13 +121,16 @@ class Pyprland: args = args[1:] full_name = f"run_{cmd}" + # Demos: + # run mako for notifications & uncomment this + # os.system(f"notify-send '{data}'") - if DEBUG: - print(f"CMD: {full_name}({args})") + self.log.debug("CMD: %s(%s)", full_name, args) await self._callHandler(full_name, *args) async def serve(self): + "Runs the server" try: async with self.server: await self.server.serve_forever() @@ -99,6 +138,7 @@ class Pyprland: await asyncio.gather(*(plugin.exit() for plugin in self.plugins.values())) async def run(self): + "Runs the server and the event listener" await asyncio.gather( asyncio.create_task(self.serve()), asyncio.create_task(self.read_events_loop()), @@ -108,25 +148,42 @@ class Pyprland: async def run_daemon(): + "Runs the server / daemon" manager = Pyprland() + err_count = itertools.count() manager.server = await asyncio.start_unix_server(manager.read_command, CONTROL) - events_reader, events_writer = await get_event_stream() + max_retry = 10 + while True: + attempt = next(err_count) + try: + events_reader, events_writer = await get_event_stream() + except Exception as e: # pylint: disable=W0718 + if attempt > max_retry: + manager.log.critical("Failed to open hyprland event stream: %s.", e) + raise PyprError() from e + manager.log.warning( + "Failed to get event stream: %s}, retry %s/%s...", e, attempt, max_retry + ) + await asyncio.sleep(1) + else: + break + 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) + except PyprError as e: + raise SystemExit(1) from e + except Exception as e: + manager.log.critical("Failed to load config.", exc_info=True) + raise SystemExit(1) from e try: await manager.run() except KeyboardInterrupt: print("Interrupted") except asyncio.CancelledError: - print("Bye!") + manager.log.critical("cancelled") finally: events_writer.close() await events_writer.wait_closed() @@ -135,20 +192,36 @@ async def run_daemon(): async def run_client(): - if sys.argv[1] == "--help": + "Runs the client (CLI)" + manager = Pyprland() + if sys.argv[1] in ("--help", "-h", "help"): + await manager.load_config(init=False) print( - """Commands: - reload - show - hide - toggle + """Syntax: pypr [command] -If arguments are ommited, runs the daemon which will start every configured command. -""" +If command is ommited, runs the daemon which will start every configured command. + +Commands: + + reload Reloads the config file (only supports adding or updating plugins)""" ) + for plug in manager.plugins.values(): + for name in dir(plug): + if name.startswith("run_"): + fn = getattr(plug, name) + if callable(fn): + print( + f" {name[4:]:20} {fn.__doc__.strip() if fn.__doc__ else 'N/A'} (from {plug.name})" + ) + return - _, writer = await asyncio.open_unix_connection(CONTROL) + try: + _, writer = await asyncio.open_unix_connection(CONTROL) + except FileNotFoundError as e: + manager.log.critical("Failed to open control socket, is pypr daemon running ?") + raise PyprError() from e + writer.write((" ".join(sys.argv[1:])).encode()) await writer.drain() writer.close() @@ -156,10 +229,25 @@ If arguments are ommited, runs the daemon which will start every configured comm def main(): + "runs the command" + if "--debug" in sys.argv: + i = sys.argv.index("--debug") + init_logger(filename=sys.argv[i + 1], force_debug=True) + del sys.argv[i : i + 2] + else: + init_logger() + ipc_init() + log = get_logger("startup") try: asyncio.run(run_daemon() if len(sys.argv) <= 1 else run_client()) except KeyboardInterrupt: pass + except PyprError: + log.critical("Command failed.") + except json.decoder.JSONDecodeError as e: + log.critical("Invalid JSON syntax in the config file: %s", e.args[0]) + except Exception: # pylint: disable=W0718 + log.critical("Unhandled exception:", exc_info=True) if __name__ == "__main__": diff --git a/pyprland/common.py b/pyprland/common.py index f09cc99..8c5947f 100644 --- a/pyprland/common.py +++ b/pyprland/common.py @@ -1,4 +1,71 @@ +""" Shared utilities: logging """ +import logging import os +__all__ = ["DEBUG", "get_logger", "init_logger"] + DEBUG = os.environ.get("DEBUG", False) -CONFIG_FILE = os.path.expanduser("~/.config/hypr/scratchpads.json") + + +class PyprError(Exception): + """Used for errors which already triggered logging""" + + +class LogObjects: + """Reusable objects for loggers""" + + handlers: list[logging.Handler] = [] + + +def init_logger(filename=None, force_debug=False): + """initializes the logging system""" + global DEBUG + if force_debug: + DEBUG = True + + class ScreenLogFormatter(logging.Formatter): + "A custom formatter, adding colors" + LOG_FORMAT = ( + r"%(name)25s - %(message)s // %(filename)s:%(lineno)d" + if DEBUG + else r"%(message)s" + ) + RESET_ANSI = "\x1b[0m" + + FORMATTERS = { + logging.DEBUG: logging.Formatter(LOG_FORMAT + RESET_ANSI), + logging.INFO: logging.Formatter(LOG_FORMAT + RESET_ANSI), + logging.WARNING: logging.Formatter("\x1b[33;20m" + LOG_FORMAT + RESET_ANSI), + logging.ERROR: logging.Formatter("\x1b[31;20m" + LOG_FORMAT + RESET_ANSI), + logging.CRITICAL: logging.Formatter("\x1b[31;1m" + LOG_FORMAT + RESET_ANSI), + } + + def format(self, record): + return self.FORMATTERS[record.levelno].format(record) + + logging.basicConfig() + if filename: + file_handler = logging.FileHandler(filename) + file_handler.setFormatter( + logging.Formatter( + fmt=r"%(asctime)s [%(levelname)s] %(name)s :: %(message)s :: %(filename)s:%(lineno)d" + ) + ) + LogObjects.handlers.append(file_handler) + stream_handler = logging.StreamHandler() + stream_handler.setFormatter(ScreenLogFormatter()) + LogObjects.handlers.append(stream_handler) + + +def get_logger(name="pypr", level=None): + "Returns a logger for `name`" + logger = logging.getLogger(name) + if level is None: + logger.setLevel(logging.DEBUG if DEBUG else logging.WARNING) + else: + logger.setLevel(level) + logger.propagate = False + for handler in LogObjects.handlers: + logger.addHandler(handler) + logger.info("Logger initialized for %s", name) + return logger diff --git a/pyprland/ipc.py b/pyprland/ipc.py index 5094231..ea269dd 100644 --- a/pyprland/ipc.py +++ b/pyprland/ipc.py @@ -1,25 +1,33 @@ #!/bin/env python +""" Interact with hyprland using sockets """ import asyncio -from typing import Any import json import os +from logging import Logger +from typing import Any -from .common import DEBUG +from .common import PyprError, get_logger +log: Logger | None = None 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(): + "Returns a new event socket connection" return await asyncio.open_unix_connection(EVENTS) async def hyprctlJSON(command) -> list[dict[str, Any]] | dict[str, Any]: """Run an IPC command and return the JSON output.""" - if DEBUG: - print("(JS)>>>", command) - ctl_reader, ctl_writer = await asyncio.open_unix_connection(HYPRCTL) + assert log + log.debug(command) + try: + ctl_reader, ctl_writer = await asyncio.open_unix_connection(HYPRCTL) + except FileNotFoundError as e: + log.critical("hyprctl socket not found! is it running ?") + raise PyprError() from e ctl_writer.write(f"-j/{command}".encode()) await ctl_writer.drain() resp = await ctl_reader.read() @@ -31,6 +39,7 @@ async def hyprctlJSON(command) -> list[dict[str, Any]] | dict[str, Any]: def _format_command(command_list, default_base_command): + "helper function to format BATCH commands" for command in command_list: if isinstance(command, str): yield f"{default_base_command} {command}" @@ -40,9 +49,14 @@ def _format_command(command_list, default_base_command): async def hyprctl(command, base_command="dispatch") -> bool: """Run an IPC command. Returns success value.""" - if DEBUG: - print(">>>", command) - ctl_reader, ctl_writer = await asyncio.open_unix_connection(HYPRCTL) + assert log + log.debug(command) + try: + ctl_reader, ctl_writer = await asyncio.open_unix_connection(HYPRCTL) + except FileNotFoundError as e: + log.critical("hyprctl socket not found! is it running ?") + raise PyprError() from e + if isinstance(command, list): ctl_writer.write( f"[[BATCH]] {' ; '.join(_format_command(command, base_command))}".encode() @@ -53,17 +67,22 @@ async def hyprctl(command, base_command="dispatch") -> bool: resp = await ctl_reader.read(100) ctl_writer.close() await ctl_writer.wait_closed() - if DEBUG: - print("<<<", resp) r: bool = resp == b"ok" * (len(resp) // 2) - if DEBUG and not r: - print(f"FAILED {resp}") + if not r: + log.error("FAILED %s", resp) return r async def get_focused_monitor_props() -> dict[str, Any]: + "Returns focused monitor data" for monitor in await hyprctlJSON("monitors"): assert isinstance(monitor, dict) - if monitor.get("focused") == True: + if monitor.get("focused"): return monitor raise RuntimeError("no focused monitor") + + +def init(): + "initialize logging" + global log + log = get_logger("ipc") diff --git a/pyprland/plugins/experimental.py b/pyprland/plugins/experimental.py index 02d6f39..b7cd2ab 100644 --- a/pyprland/plugins/experimental.py +++ b/pyprland/plugins/experimental.py @@ -1,7 +1,8 @@ +" Plugin template " from .interface import Plugin -from ..ipc import hyprctlJSON, hyprctl +# from ..ipc import hyprctlJSON, hyprctl class Extension(Plugin): - pass + "Sample plugin template" diff --git a/pyprland/plugins/expose.py b/pyprland/plugins/expose.py new file mode 100644 index 0000000..ed01925 --- /dev/null +++ b/pyprland/plugins/expose.py @@ -0,0 +1,54 @@ +""" expose Brings every client window to screen for selection +toggle_minimized allows having an "expose" like selection of minimized windows +""" +from typing import Any, cast + +from ..ipc import hyprctl, hyprctlJSON +from .interface import Plugin + + +class Extension(Plugin): # pylint: disable=missing-class-docstring + exposed: list[dict] = [] + + async def run_toggle_minimized(self, special_workspace="minimized"): + """[name] Toggles switching the focused window to the special workspace "name" (default: minimized)""" + aw = cast(dict, await hyprctlJSON("activewindow")) + wid = aw["workspace"]["id"] + assert isinstance(wid, int) + if wid < 1: # special workspace: unminimize + wrk = cast(dict, await hyprctlJSON("activeworkspace")) + await hyprctl(f"togglespecialworkspace {special_workspace}") + await hyprctl(f"movetoworkspacesilent {wrk['id']},address:{aw['address']}") + await hyprctl(f"focuswindow address:{aw['address']}") + else: + await hyprctl( + f"movetoworkspacesilent special:{special_workspace},address:{aw['address']}" + ) + + @property + def exposed_clients(self): + "Returns the list of clients currently using exposed mode" + if self.config.get("include_special", False): + return self.exposed + return [c for c in self.exposed if c["workspace"]["id"] > 0] + + async def run_expose(self): + """Expose every client on the active workspace. + If expose is active restores everything and move to the focused window""" + if self.exposed: + aw: dict[str, Any] = cast(dict, await hyprctlJSON("activewindow")) + focused_addr = aw["address"] + for client in self.exposed_clients: + await hyprctl( + f"movetoworkspacesilent {client['workspace']['id']},address:{client['address']}" + ) + # await hyprctl("togglespecialworkspace exposed") + await hyprctl(f"focuswindow address:{focused_addr}") + self.exposed = [] + else: + self.exposed = cast(list, await hyprctlJSON("clients")) + for client in self.exposed_clients: + await hyprctl( + f"movetoworkspacesilent special:exposed,address:{client['address']}" + ) + await hyprctl("togglespecialworkspace exposed") diff --git a/pyprland/plugins/interface.py b/pyprland/plugins/interface.py index fb31597..4613695 100644 --- a/pyprland/plugins/interface.py +++ b/pyprland/plugins/interface.py @@ -1,17 +1,26 @@ +" Common plugin interface " from typing import Any +from ..common import get_logger + class Plugin: + "Base plugin class, handles logger and config" + def __init__(self, name: str): + "create a new plugin `name` and the matching logger" self.name = name + self.log = get_logger(name) + self.config: dict[str, Any] = {} async def init(self): - pass + "empty init function" async def exit(self): - return + "empty exit function" async def load_config(self, config: dict[str, Any]): + "Loads the configuration section from the passed `config`" try: self.config = config[self.name] except KeyError: diff --git a/pyprland/plugins/ironbar.py b/pyprland/plugins/ironbar.py new file mode 100644 index 0000000..219d0e8 --- /dev/null +++ b/pyprland/plugins/ironbar.py @@ -0,0 +1,27 @@ +" 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) diff --git a/pyprland/plugins/lost_windows.py b/pyprland/plugins/lost_windows.py new file mode 100644 index 0000000..4556dfa --- /dev/null +++ b/pyprland/plugins/lost_windows.py @@ -0,0 +1,45 @@ +" Moves unreachable client windows to the currently focused workspace" +from typing import Any, cast + +from ..ipc import hyprctl, hyprctlJSON +from .interface import Plugin + + +def contains(monitor, window): + "Tell if a window is visible in a monitor" + if not ( + window["at"][0] > monitor["x"] + and window["at"][0] < monitor["x"] + monitor["width"] + ): + return False + if not ( + window["at"][1] > monitor["y"] + and window["at"][1] < monitor["y"] + monitor["height"] + ): + return False + return True + + +class Extension(Plugin): # pylint: disable=missing-class-docstring + async def run_attract_lost(self): + """Brings lost floating windows to the current workspace""" + monitors = cast(list, await hyprctlJSON("monitors")) + windows = cast(list, await hyprctlJSON("clients")) + lost = [ + win + for win in windows + if win["floating"] and not any(contains(mon, win) for mon in monitors) + ] + focused: dict[str, Any] = [mon for mon in monitors if mon["focused"]][0] + interval = focused["width"] / (1 + len(lost)) + interval_y = focused["height"] / (1 + len(lost)) + batch = [] + workspace: int = focused["activeWorkspace"]["id"] + margin = interval // 2 + margin_y = interval_y // 2 + for i, window in enumerate(lost): + pos_x = int(margin + focused["x"] + i * interval) + pos_y = {int(margin_y + focused["y"] + i * interval_y)} + batch.append(f'movetoworkspacesilent {workspace},pid:{window["pid"]}') + batch.append(f'movewindowpixel exact {pos_x} {pos_y},pid:{window["pid"]}') + await hyprctl(batch) diff --git a/pyprland/plugins/magnify.py b/pyprland/plugins/magnify.py new file mode 100644 index 0000000..fe3a296 --- /dev/null +++ b/pyprland/plugins/magnify.py @@ -0,0 +1,21 @@ +" Toggles workspace zooming " +from ..ipc import hyprctl +from .interface import Plugin + + +class Extension(Plugin): # pylint: disable=missing-class-docstring + zoomed = False + + async def run_zoom(self, *args): + """[factor] zooms to "factor" or toggles zoom level ommited""" + if args: + value = int(args[0]) + await hyprctl(f"misc:cursor_zoom_factor {value}", "keyword") + self.zoomed = value != 1 + else: # toggle + if self.zoomed: + await hyprctl("misc:cursor_zoom_factor 1", "keyword") + else: + fact = int(self.config.get("factor", 2)) + await hyprctl(f"misc:cursor_zoom_factor {fact}", "keyword") + self.zoomed = not self.zoomed diff --git a/pyprland/plugins/monitors.py b/pyprland/plugins/monitors.py index f8e0918..18556ed 100644 --- a/pyprland/plugins/monitors.py +++ b/pyprland/plugins/monitors.py @@ -1,45 +1,111 @@ -from typing import Any -from .interface import Plugin +" The monitors plugin " import subprocess +from typing import Any, cast from ..ipc import hyprctlJSON +from .interface import Plugin -class Extension(Plugin): - async def event_monitoradded(self, screenid): - screenid = screenid.strip() +def configure_monitors(monitors, screenid: str, pos_x: int, pos_y: int) -> None: + "Apply the configuration change" + x_offset = -pos_x if pos_x < 0 else 0 + y_offset = -pos_y if pos_y < 0 else 0 + + min_x = pos_x + min_y = pos_y + + command = ["wlr-randr"] + other_monitors = [mon for mon in monitors if mon["name"] != screenid] + for mon in other_monitors: + min_x = min(min_x, mon["x"]) + min_y = min(min_y, mon["y"]) + x_offset = -min_x + y_offset = -min_y + for mon in other_monitors: + command.extend( + [ + "--output", + mon["name"], + "--pos", + f"{mon['x']+x_offset},{mon['y']+y_offset}", + ] + ) + + command.extend( + ["--output", screenid, "--pos", f"{pos_x+x_offset},{pos_y+y_offset}"] + ) + subprocess.call(command) + + +class Extension(Plugin): # pylint: disable=missing-class-docstring + async def load_config(self, config) -> None: + await super().load_config(config) + await self.run_relayout() + + async def run_relayout(self): + monitors = cast(list[dict], await hyprctlJSON("monitors")) + for monitor in monitors: + await self.event_monitoradded( + monitor["name"], no_default=True, monitors=monitors + ) + + async def event_monitoradded( + self, monitor_name, no_default=False, monitors: list | None = None + ) -> None: + "Triggers when a monitor is plugged" + monitor_name = monitor_name.strip() + + if not monitors: + monitors = cast(list, await hyprctlJSON("monitors")) + + assert monitors - monitors: list[dict[str, Any]] = await hyprctlJSON("monitors") for mon in monitors: - if mon["name"].startswith(screenid): - mon_name = mon["description"] + if mon["name"].startswith(monitor_name): + mon_description = mon["description"] break else: - print(f"Monitor {screenid} not found") + self.log.info("Monitor %s not found", monitor_name) return + if self._place_monitors(monitor_name, mon_description, monitors): + return + + if not no_default: + default_command = self.config.get("unknown") + if default_command: + subprocess.call(default_command, shell=True) + + def _place_monitors( + self, monitor_name: str, mon_description: str, monitors: list[dict[str, Any]] + ): + "place a given monitor according to config" mon_by_name = {m["name"]: m for m in monitors} - - newmon = mon_by_name[screenid] - + newmon = mon_by_name[monitor_name] 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 mon_pattern in mon_description: + for placement, other_mon_description in conf.items(): + try: + ref = mon_by_name[other_mon_description] + except KeyError: + continue if ref: place = placement.lower() + x: int = 0 + y: int = 0 if place == "topof": - x: int = ref["x"] - y: int = ref["y"] - newmon["height"] + x = ref["x"] + y = ref["y"] - newmon["height"] elif place == "bottomof": - x: int = ref["x"] - y: int = ref["y"] + ref["height"] + x = ref["x"] + y = ref["y"] + ref["height"] elif place == "leftof": - x: int = ref["x"] - newmon["width"] - y: int = ref["y"] + x = ref["x"] - newmon["width"] + y = ref["y"] else: # rightof - x: int = ref["x"] + ref["width"] - y: int = ref["y"] - subprocess.call( - ["wlr-randr", "--output", screenid, "--pos", f"{x},{y}"] - ) + x = ref["x"] + ref["width"] + y = ref["y"] + + configure_monitors(monitors, monitor_name, x, y) + return True + return False diff --git a/pyprland/plugins/scratchpads.py b/pyprland/plugins/scratchpads.py index e203cee..df1d466 100644 --- a/pyprland/plugins/scratchpads.py +++ b/pyprland/plugins/scratchpads.py @@ -1,42 +1,50 @@ -import subprocess -from typing import Any +" Scratchpads addon " import asyncio -from ..ipc import ( - hyprctl, - hyprctlJSON, - get_focused_monitor_props, -) import os +from itertools import count +import subprocess +from typing import Any, cast +import logging +from ..ipc import get_focused_monitor_props, hyprctl, hyprctlJSON from .interface import Plugin DEFAULT_MARGIN = 60 -async def get_client_props_by_pid(pid: int): +async def get_client_props_by_address(addr: str): + "Returns client properties given its address" + assert len(addr) > 2, "Client address is invalid" for client in await hyprctlJSON("clients"): assert isinstance(client, dict) - if client.get("pid") == pid: + if client.get("address") == addr: return client class Animations: - @classmethod - async def fromtop(cls, monitor, client, client_uid, margin): + "Animation store" + + @staticmethod + async def fromtop(monitor, client, client_uid, margin): + "Slide from/to top" + scale = float(monitor["scale"]) mon_x = monitor["x"] mon_y = monitor["y"] - mon_width = monitor["width"] + mon_width = int(monitor["width"] / scale) 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}") - @classmethod - async def frombottom(cls, monitor, client, client_uid, margin): + @staticmethod + async def frombottom(monitor, client, client_uid, margin): + "Slide from/to bottom" + scale = float(monitor["scale"]) mon_x = monitor["x"] mon_y = monitor["y"] - mon_width = monitor["width"] - mon_height = monitor["height"] + mon_width = int(monitor["width"] / scale) + mon_height = int(monitor["height"] / scale) client_width = client["size"][0] client_height = client["size"][1] @@ -45,23 +53,27 @@ class Animations: f"movewindowpixel exact {margin_x} {mon_y + mon_height - client_height - margin},{client_uid}" ) - @classmethod - async def fromleft(cls, monitor, client, client_uid, margin): + @staticmethod + async def fromleft(monitor, client, client_uid, margin): + "Slide from/to left" + scale = float(monitor["scale"]) mon_x = monitor["x"] mon_y = monitor["y"] - mon_height = monitor["height"] + mon_height = int(monitor["height"] / scale) client_height = client["size"][1] margin_y = int((mon_height - client_height) / 2) + mon_y await hyprctl(f"movewindowpixel exact {margin + mon_x} {margin_y},{client_uid}") - @classmethod - async def fromright(cls, monitor, client, client_uid, margin): + @staticmethod + async def fromright(monitor, client, client_uid, margin): + "Slide from/to right" + scale = float(monitor["scale"]) mon_x = monitor["x"] mon_y = monitor["y"] - mon_width = monitor["width"] - mon_height = monitor["height"] + mon_width = int(monitor["width"] / scale) + mon_height = int(monitor["height"] / scale) client_width = client["size"][0] client_height = client["size"][1] @@ -72,54 +84,75 @@ class Animations: class Scratch: + "A scratchpad state including configuration & client state" + log = logging.getLogger("scratch") + def __init__(self, uid, opts): self.uid = uid self.pid = 0 self.conf = opts self.visible = False self.just_created = True - self.clientInfo = {} + self.client_info = {} def isAlive(self) -> bool: + "is the process running ?" 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)" + with open(os.path.join(path, "status"), "r", encoding="utf-8") as f: + for line in f.readlines(): + if line.startswith("State"): + state = line.split()[1] + return state not in "ZX" # not "Z (zombie)"or "X (dead)" return False def reset(self, pid: int) -> None: + "clear the object" self.pid = pid self.visible = False self.just_created = True - self.clientInfo = {} + self.client_info = {} @property def address(self) -> str: - return str(self.clientInfo.get("address", ""))[2:] + "Returns the client address" + return str(self.client_info.get("address", ""))[2:] - async def updateClientInfo(self, clientInfo=None) -> None: - if clientInfo is None: - clientInfo = await get_client_props_by_pid(self.pid) - assert isinstance(clientInfo, dict) - self.clientInfo.update(clientInfo) + async def updateClientInfo(self, client_info=None) -> None: + "update the internal client info property, if not provided, refresh based on the current address" + if client_info is None: + client_info = await get_client_props_by_address("0x" + self.address) + try: + 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) + + def __str__(self): + return f"{self.uid} {self.address} : {self.client_info} / {self.conf}" -class Extension(Plugin): - async def init(self) -> None: - 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] = {} +class Extension(Plugin): # pylint: disable=missing-class-docstring + procs: dict[str, subprocess.Popen] = {} + scratches: dict[str, Scratch] = {} + transitioning_scratches: set[str] = set() + _new_scratches: set[str] = set() + _respawned_scratches: set[str] = set() + scratches_by_address: dict[str, Scratch] = {} + scratches_by_pid: dict[int, Scratch] = {} + focused_window_tracking: dict[str, dict] = {} async def exit(self) -> None: + "exit hook" + async def die_in_piece(scratch: Scratch): proc = self.procs[scratch.uid] proc.terminate() - for n in range(10): + for _ in range(10): if not scratch.isAlive(): break await asyncio.sleep(0.1) @@ -131,9 +164,10 @@ class Extension(Plugin): *(die_in_piece(scratch) for scratch in self.scratches.values()) ) - async def load_config(self, config) -> None: - config: dict[str, dict[str, Any]] = config["scratchpads"] - scratches = {k: Scratch(k, v) for k, v in config.items()} + async def load_config(self, config: dict[str, Any]) -> None: + "config loader" + my_config: dict[str, dict[str, Any]] = config["scratchpads"] + scratches = {k: Scratch(k, v) for k, v in my_config.items()} new_scratches = set() @@ -146,139 +180,215 @@ class Extension(Plugin): # not known yet for name in new_scratches: - self.start_scratch_command(name) + if not self.scratches[name].conf.get("lazy", False): + await self.start_scratch_command(name, is_new=True) - def start_scratch_command(self, name: str) -> None: + async def start_scratch_command(self, name: str, is_new=False) -> None: + "spawns a given scratchpad's process" + if is_new: + self._new_scratches.add(name) 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( + proc = subprocess.Popen( scratch.conf["command"], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, shell=True, ) - pid = self.procs[name].pid + self.procs[name] = proc + pid = proc.pid self.scratches[name].reset(pid) - self.scratches_by_pid[self.procs[name].pid] = scratch - if old_pid: + self.scratches_by_pid[pid] = scratch + self.log.info(f"scratch {scratch.uid} has pid {pid}") + + if old_pid and old_pid in self.scratches_by_pid: del self.scratches_by_pid[old_pid] # Events async def event_activewindowv2(self, addr) -> None: + "active windows hook" addr = addr.strip() scratch = self.scratches_by_address.get(addr) if scratch: if scratch.just_created: + self.log.debug("Hiding just created scratch %s", scratch.uid) 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: + self.log.info((scratch.address, addr)) + if scratch.client_info 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) + self.log.debug("hide %s because another client is active", uid) + await self.run_hide(uid, autohide=True) + + async def _alternative_lookup(self): + "if class attribute is defined, use class matching and return True" + class_lookup_hack = [ + self.scratches[name] + for name in self._respawned_scratches + if self.scratches[name].conf.get("class") + ] + if not class_lookup_hack: + return False + self.log.debug("Lookup hack triggered") + # hack to update the client info from the provided class + for client in await hyprctlJSON("clients"): + assert isinstance(client, dict) + for pending_scratch in class_lookup_hack: + if pending_scratch.conf["class"] == client["class"]: + self.scratches_by_address[client["address"][2:]] = pending_scratch + self.log.debug("client class found: %s", client) + await pending_scratch.updateClientInfo(client) + return True async def event_openwindow(self, params) -> None: - addr, wrkspc, kls, title = params.split(",", 3) - if wrkspc.startswith("special"): + "open windows hook" + addr, wrkspc, _kls, _title = params.split(",", 3) + if self._respawned_scratches: item = self.scratches_by_address.get(addr) if not item and self._respawned_scratches: - await self.updateScratchInfo() + # hack for windows which aren't related to the process (see #8) + if not await self._alternative_lookup(): + await self.updateScratchInfo() item = self.scratches_by_address.get(addr) if item and item.just_created: + if item.uid in self._new_scratches: + await self.run_hide(item.uid, force=True) + self._new_scratches.discard(item.uid) 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) -> None: + """ toggles visibility of scratchpad "name" """ uid = uid.strip() item = self.scratches.get(uid) if not item: - print(f"{uid} is not configured") + self.log.warning("%s is not configured", uid) return + self.log.debug("%s is visible = %s", uid, item.visible) if item.visible: await self.run_hide(uid) else: await self.run_show(uid) - async def updateScratchInfo(self, scratch: Scratch | None = 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) + async def _anim_hide(self, animation_type, scratch): + "animate hiding a scratchpad" + addr = "address:0x" + scratch.address + offset = scratch.conf.get("offset") + if offset is None: + if "size" not in scratch.client_info: + await self.updateScratchInfo(scratch) + + offset = int(1.3 * scratch.client_info["size"][1]) + + if animation_type == "fromtop": + await hyprctl(f"movewindowpixel 0 -{offset},{addr}") + elif animation_type == "frombottom": + await hyprctl(f"movewindowpixel 0 {offset},{addr}") + elif animation_type == "fromleft": + await hyprctl(f"movewindowpixel -{offset} 0,{addr}") + elif animation_type == "fromright": + await hyprctl(f"movewindowpixel {offset} 0,{addr}") + + if scratch.uid in self.transitioning_scratches: + return # abort sequence + await asyncio.sleep(0.2) # await for animation to finish + + async def updateScratchInfo(self, orig_scratch: Scratch | None = None) -> None: + """Update every scratchpads information if no `scratch` given, + else update a specific scratchpad info""" + pid = orig_scratch.pid if orig_scratch else None + for client in await hyprctlJSON("clients"): + assert isinstance(client, dict) + if pid and pid != client["pid"]: + continue + scratch = self.scratches_by_address.get(client["address"][2:]) + if not scratch: + scratch = self.scratches_by_pid.get(client["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 + self.scratches_by_address[client["address"][2:]] = scratch + if scratch: + await scratch.updateClientInfo(client) - async def run_hide(self, uid: str, force=False) -> None: + async def run_hide(self, uid: str, force=False, autohide=False) -> None: + """ hides scratchpad "name" """ uid = uid.strip() - item = self.scratches.get(uid) - if not item: - print(f"{uid} is not configured") + scratch = self.scratches.get(uid) + if not scratch: + self.log.warning("%s is not configured", uid) return - if not item.visible and not force: - print(f"{uid} is already hidden") + if not scratch.visible and not force: + self.log.warning("%s is already hidden", uid) return - item.visible = False - pid = "pid:%d" % item.pid - animation_type: str = item.conf.get("animation", "").lower() + scratch.visible = False + if not scratch.isAlive(): + await self.run_show(uid, force=True) + return + self.log.info("Hiding %s", uid) + addr = "address:0x" + scratch.address + animation_type: str = scratch.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) + await self._anim_hide(animation_type, scratch) - 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}") + await hyprctl(f"movetoworkspacesilent special:scratch_{uid},{addr}") + + if ( + animation_type and uid in self.focused_window_tracking + ): # focus got lost when animating + if not autohide and "address" in self.focused_window_tracking[uid]: + await hyprctl( + f"focuswindow address:{self.focused_window_tracking[uid]['address']}" + ) + 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: + """ shows scratchpad "name" """ uid = uid.strip() item = self.scratches.get(uid) + self.focused_window_tracking[uid] = cast( + dict[str, Any], await hyprctlJSON("activewindow") + ) + if not item: - print(f"{uid} is not configured") + self.log.warning("%s is not configured", uid) return if item.visible and not force: - print(f"{uid} is already visible") + self.log.warning("%s is already visible", uid) 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) + self.log.info("Showing %s", uid) + await self.ensure_alive(uid, item) item.visible = True monitor = await get_focused_monitor_props() @@ -286,20 +396,64 @@ class Extension(Plugin): await self.updateScratchInfo(item) - pid = "pid:%d" % item.pid + assert item.address, "No address !" + + addr = "address:0x" + item.address animation_type = item.conf.get("animation", "").lower() wrkspc = monitor["activeWorkspace"]["id"] self.transitioning_scratches.add(uid) - await hyprctl(f"moveworkspacetomonitor special:scratch {monitor['name']}") - await hyprctl(f"movetoworkspacesilent {wrkspc},{pid}") + await hyprctl(f"moveworkspacetomonitor special:scratch_{uid} {monitor['name']}") + await hyprctl(f"movetoworkspacesilent {wrkspc},{addr}") if animation_type: margin = item.conf.get("margin", DEFAULT_MARGIN) fn = getattr(Animations, animation_type) - await fn(monitor, item.clientInfo, pid, margin) + await fn(monitor, item.client_info, addr, margin) + + 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 hyprctl(f"focuswindow {pid}") await asyncio.sleep(0.2) # ensure some time for events to propagate 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 diff --git a/pyprland/plugins/shift_monitors.py b/pyprland/plugins/shift_monitors.py new file mode 100644 index 0000000..2e6d91c --- /dev/null +++ b/pyprland/plugins/shift_monitors.py @@ -0,0 +1,33 @@ +" shift workspaces across monitors " +from typing import cast + +from ..ipc import hyprctl, hyprctlJSON +from .interface import Plugin + + +class Extension(Plugin): # pylint: disable=missing-class-docstring + monitors: list[str] = [] + + async def init(self): + self.monitors: list[str] = [ + mon["name"] for mon in cast(list[dict], await hyprctlJSON("monitors")) + ] + + async def run_shift_monitors(self, arg: str): + """Swaps monitors' workspaces in the given direction""" + direction: int = int(arg) + if direction > 0: + mon_list = self.monitors[:-1] + else: + mon_list = list(reversed(self.monitors[1:])) + + for i, mon in enumerate(mon_list): + await hyprctl(f"swapactiveworkspaces {mon} {self.monitors[i+direction]}") + + async def event_monitoradded(self, monitor): + "keep track of monitors" + self.monitors.append(monitor.strip()) + + async def event_monitorremoved(self, monitor): + "keep track of monitors" + self.monitors.remove(monitor.strip()) diff --git a/pyprland/plugins/toggle_dpms.py b/pyprland/plugins/toggle_dpms.py new file mode 100644 index 0000000..a395519 --- /dev/null +++ b/pyprland/plugins/toggle_dpms.py @@ -0,0 +1,16 @@ +" Toggle monitors on or off " +from typing import Any, cast + +from ..ipc import hyprctl, hyprctlJSON +from .interface import Plugin + + +class Extension(Plugin): # pylint: disable=missing-class-docstring + async def run_toggle_dpms(self): + """toggles dpms on/off for every monitor""" + monitors = cast(list[dict[str, Any]], await hyprctlJSON("monitors")) + powered_off = any(m["dpmsStatus"] for m in monitors) + if not powered_off: + await hyprctl("dpms on") + else: + await hyprctl("dpms off") diff --git a/pyprland/plugins/workspaces_follow_focus.py b/pyprland/plugins/workspaces_follow_focus.py index d60dc34..24efe8b 100644 --- a/pyprland/plugins/workspaces_follow_focus.py +++ b/pyprland/plugins/workspaces_follow_focus.py @@ -1,36 +1,43 @@ -import asyncio +""" Force workspaces to follow the focus / mouse """ +from typing import cast + +from ..ipc import hyprctl, hyprctlJSON from .interface import Plugin -from ..ipc import hyprctlJSON, hyprctl +class Extension(Plugin): # pylint: disable=missing-class-docstring + workspace_list: list[int] = [] -class Extension(Plugin): async def load_config(self, config): + "loads the config" await super().load_config(config) - self.workspace_list = list(range(1, self.config.get("max_workspaces", 10))) + self.workspace_list = list(range(1, self.config.get("max_workspaces", 10) + 1)) async def event_focusedmon(self, screenid_index): + "reacts to monitor changes" monitor_id, workspace_id = screenid_index.split(",") workspace_id = int(workspace_id) # move every free workspace to the currently focused desktop busy_workspaces = set( mon["activeWorkspace"]["id"] - for mon in await hyprctlJSON("monitors") + for mon in cast(list[dict], await hyprctlJSON("monitors")) if mon["name"] != monitor_id ) - workspaces = [w["id"] for w in await hyprctlJSON("workspaces") if w["id"] > 0] + workspaces = [ + w["id"] + for w in cast(list[dict], await hyprctlJSON("workspaces")) + if w["id"] > 0 + ] - batch: list[str | list[str]] = [["animations:enabled false", "keyword"]] + batch: list[str | list[str]] = [] for n in workspaces: if n in busy_workspaces or n == workspace_id: continue batch.append(f"moveworkspacetomonitor {n} {monitor_id}") - batch.append(f"workspace {workspace_id}") await hyprctl(batch) - await asyncio.sleep(0.05) - await hyprctl("animations:enabled true", base_command="keyword") async def run_change_workspace(self, direction: str): + """<+1/-1> Switch workspaces of current monitor, avoiding displayed workspaces""" increment = int(direction) # get focused screen info monitors = await hyprctlJSON("monitors") @@ -38,21 +45,24 @@ class Extension(Plugin): for monitor in monitors: if monitor["focused"]: break + else: + self.log.error("Can not find a focused monitor") + return assert isinstance(monitor, dict) busy_workspaces = set( m["activeWorkspace"]["id"] for m in monitors if m["id"] != monitor["id"] ) - # get workspaces info - workspaces = await hyprctlJSON("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) - ] + try: + idx = available_workspaces.index(cur_workspace) + except ValueError: + next_workspace = available_workspaces[0 if increment > 0 else -1] + else: + next_workspace = available_workspaces[ + (idx + increment) % len(available_workspaces) + ] await hyprctl(f"moveworkspacetomonitor {next_workspace},{monitor['name']}") await hyprctl(f"workspace {next_workspace}") diff --git a/pyproject.toml b/pyproject.toml index f58b539..a6adc29 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,12 +1,12 @@ [tool.poetry] name = "pyprland" -version = "1.0.2" +version = "1.4.1" description = "An hyperland plugin system" authors = ["fdev31 "] license = "MIT" readme = "README.md" packages = [{include = "pyprland"}] -homepage = "https://github.com/fdev31/pyprland/" +homepage = "https://github.com/hyprland-community/pyprland/" [tool.poetry.scripts] pypr = "pyprland.command:main"