expose: disable togglespecialworkspace when hiding the workspace because movetoworkspacesilent is already doing it

Scratchpads: Dynamic position and size, depending on monitor

.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')

.gitignore

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

.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
 

.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 <file or directory> 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=

README.md

@@ -4,326 +4,59 @@
 
 Host process for multiple Hyprland plugins.
 
-- **tool**: `pypr`
+Check the [wiki](https://github.com/hyprland-community/pyprland/wiki) for more information.
-- **config file**: `~/.config/hypr/pyprland.json`
 
-The `pypr` tool only have two built-in commands:
+# 1.4.2 (WIP)
 
-- `reload` reads the configuration file and attempt to apply the changes
+- [two new options](https://github.com/hyprland-community/pyprland/wiki/Plugins#size-optional) for scratchpads: `position` and `size` - from @iliayar
-- `--help` lists available commands (including plugins commands)
+- bugfixes
 
-Other commands are added by adding plugins.
+# 1.4.1
 
-A single config file `~/.config/hypr/pyprland.json` is used, using the following syntax:
+- minor bugfixes
 
-```json
+# 1.4.0
-{
-  "pyprland": {
-    "plugins": ["plugin_name"]
-  },
-  "plugin_name": {
-    "plugin_option": 42
-  }
-}
-```
 
-## Built-in plugins
+- 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 <filename>`
 
-- `scratchpads` implements dropdowns & togglable poppups
+## 1.3.1
-- `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. If you think the multi-screen behavior of hyprland is not usable or broken/unexpected, this is probably for you.
-- `lost_windows` brings lost floating windows to the current workspace
-- `toggle_dpms` toggles the DPMS status of every plugged monitor
-- `magnify` toggles zooming of viewport or sets a specific scaling factor
-- `shift_monitors` adds a self-configured "swapactiveworkspaces" command
 
-## Installation
+- `monitors` triggers rules on startup (not only when a monitor is plugged)
 
-Use the python package manager:
+## 1.3.0
 
-```
+- Add `shift_monitors` addon
-pip install pyprland
+- Add `monitors` addon
-```
+- scratchpads: more reliable client tracking
+- bugfixes
 
-If you run archlinux, you can also find it on AUR: `yay -S pyprland`
+## 1.2.1
 
-Don't forget to start the process with hyprland, adding to `hyprland.conf`:
+- scratchpads have their own special workspaces now
+- misc improvements
 
-```
+## 1.2.0
-exec-once = pypr
-```
 
-## Getting started
+- Add `magnify` addon
+- focus fix when closing a scratchpad
+- misc improvements
 
-Create a configuration file in `~/.config/hypr/pyprland.json` enabling a list of plugins, each plugin may have its own configuration needs, eg:
+## 1.1.0
 
-```json
+- Add `lost_windows` addon
-{
+- Add `toggle_dpms` addon
-  "pyprland": {
+- `workspaces_follow_focus` now requires hyprland 0.25.0
-    "plugins": [
+- misc improvements
-      "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"
-      }
-    }
-    "unknown": "wlrlui"
-  }
-}
-```
 
-# Plugin: `shift_monitors`
+## 1.0.1, 1.0.2
 
-Swaps the workspaces of every screen in the given direction.
+- bugfixes & improvements
-Note the behavior can be hard to predict if you have more than 2 monitors, suggestions are welcome.
 
-### Command
+## 1.0
 
-- `shift_monitors <direction>`: swaps every monitor's workspace in the given direction
+- First release, a modular hpr-scratcher (`scratchpads` plugin)
-
+- Add `workspaces_follow_focus` addon
-Example usage in `hyprland.conf`:
-
-```
-bind = $mainMod SHIFT, O, exec, pypr shift_monitors +1
- ```
-
-# Plugin: `magnify`
-
-### Command
-
-- `zoom [value]`: if no value, toggles magnification. If an integer is provided, it will set as scaling factor.
-
-### Configuration
-
-
-#### `factor` (optional, defaults to 2)
-
-Scaling factor to be used when no value is provided.
-
-# Plugin: `toggle_dpms`
-
-### Command
-
-- `toggle_dpms`: if any screen is powered on, turn them all off, else turn them all on
-
-
-# Plugin: `lost_windows`
-
-### Command
-
-- `attract_lost`: brings the lost windows to the current screen / workspace
-
-# Plugin: `monitors`
-
-Syntax:
-```json
-"monitors": {
-  "placement": {
-    "<partial model description>": {
-      "placement type": "<monitor name/output>"
-    },
-    "unknown": "<command to run for unknown monitors>"
-  }
-}
-```
-
-Example:
-```json
-"monitors": {
-  "unknown": "notify-send 'Unknown monitor'",
-  "placement": {
-    "Sony": {
-      "topOf": "HDMI-1"
-    }
-  }
-}
-```
-
-Requires `wlr-randr`.
-
-Allows relative placement of monitors depending on the model ("description" returned by `hyprctl monitors`).
-
-### Configuration
-
-
-#### `placement`
-
-Supported placements are:
-
-- leftOf
-- topOf
-- rightOf
-- bottomOf
-
-#### `unknown` (optional)
-
-If set, runs the associated command for screens which aren't matching any of the provided placements (pattern isn't found in monitor description).
-
-**Note** this is supposed to be a short lived command which will block the rest of the process until closed. In other words no plugin will be processed while this command remains open.
-
-# Plugin: `workspaces_follow_focus`
-
-Make non-visible workspaces follow the focused monitor.
-Also provides commands to switch between workspaces wile preserving the current monitor assignments: 
-
-Syntax:
-```json
-"workspaces_follow_focus": {
-  "max_workspaces": <number of workspaces>
-}
-```
-
-### Command
-
-- `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`.
-
-# Plugin: `scratchpads`
-
-Defines commands that should run in dropdowns. Successor of [hpr-scratcher](https://github.com/hyprland-community/hpr-scratcher), it's fully compatible, just put the configuration under "scratchpads".
-
-Syntax:
-```json
-"scratchpads": {
-  "scratchpad name": {
-    "command": "command to run"
-  }
-}
-```
-
-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
-
-Example:
-```json
-"scratchpads": {
-  "term": {
-    "command": "kitty --class kitty-dropterm",
-    "animation": "fromTop",
-    "margin": 50,
-    "unfocus": "hide"
-  },
-  "volume": {
-    "command": "pavucontrol",
-    "animation": "fromRight"
-  }
-}
-```
-
-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
-```
-
-And you'll be able to toggle pavucontrol with MOD + V.
-
-### Commands
-
-- `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)
-
-
-
-### Configuration
-
-#### `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` (optional)
-
-Type of animation to use
-
-- `null` / `""` / not defined (no animation)
-- "fromTop" (stays close to top screen border)
-- "fromBottom" (stays close to bottom screen border)
-- "fromLeft" (stays close to left screen border)
-- "fromRight" (stays close to right screen border)
-
-#### `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
-
-# Writing plugins
-
-You can start enabling a plugin called "experimental" and add code to `plugins/experimental.py`.
-A better way is to copy this as a starting point and make your own python module.
-Plugins can be loaded with full python module path, eg: `"mymodule.pyprlandplugin"`, the loaded module must provide an `Extension` interface.
-
-Check the `interface.py` file to know the base methods, also have a look at the other plugins for working examples.
-
-To get more details when an error is occurring, `export DEBUG=1` in your shell before running.
-
-## Creating a command
-
-Just add a method called `run_<name of your command>`, eg with "togglezoom" command:
-
-```python
-async def init(self):
-  self.zoomed = False
-
-async def run_togglezoom(self, args):
-  if self.zoomed:
-    await hyprctl('misc:cursor_zoom_factor 1', 'keyword')
-  else:
-    await hyprctl('misc:cursor_zoom_factor 2', 'keyword')
-  self.zoomed = not self.zoomed
-```
-
-## Reacting to an event
-
-Similar as a command, implement some `event_<the event you are interested in>` method.
 

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
+}

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;
+    };
+  }
+  );
+}

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"

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,19 +19,33 @@ 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, init=True):
-        self.config = json.loads(
+        """Loads the configuration
-            open(os.path.expanduser(CONFIG_FILE), encoding="utf-8").read()
+
-        )
+        if `init` is true, also initializes the plugins"""
-        for name in self.config["pyprland"]["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:
@@ -38,38 +54,57 @@ class Pyprland:
                         await plug.init()
                     self.plugins[name] = plug
                 except Exception as e:
-                    print(f"Error loading plugin {name}: {e}")
+                    self.log.error("Error loading plugin %s:", name, exc_info=True)
-                    if DEBUG:
+                    raise PyprError() from e
-                        traceback.print_exc()
             if init:
-                await self.plugins[name].load_config(self.config)
+                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:
+                except AssertionError as e:
-                    print(f"{plugin.name}::{full_name}({params}) failed:")
+                    self.log.error(
-                    traceback.print_exc()
+                        "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
@@ -86,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:
+        self.log.debug("CMD: %s(%s)", full_name, args)
-            print(f"CMD: {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()
@@ -100,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()),
@@ -109,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:
+    except PyprError as e:
-        print(
+        raise SystemExit(1) from e
-            f"No config file found, create one at {CONFIG_FILE} with a valid pyprland.plugins list"
+    except Exception as e:
-        )
+        manager.log.critical("Failed to load config.", exc_info=True)
-        raise SystemExit(1)
+        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()
@@ -136,8 +192,9 @@ async def run_daemon():
 
 
 async def run_client():
-    if sys.argv[1] in ("--help", "-h"):
+    "Runs the client (CLI)"
-        manager = Pyprland()
+    manager = Pyprland()
+    if sys.argv[1] in ("--help", "-h", "help"):
         await manager.load_config(init=False)
         print(
             """Syntax: pypr [command]
@@ -159,7 +216,12 @@ Commands:
 
         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()
@@ -167,10 +229,25 @@ Commands:
 
 
 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__":

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

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:
+    assert log
-        print("(JS)>>>", command)
+    log.debug(command)
-    ctl_reader, ctl_writer = await asyncio.open_unix_connection(HYPRCTL)
+    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:
+    assert log
-        print(">>>", command)
+    log.debug(command)
-    ctl_reader, ctl_writer = await asyncio.open_unix_connection(HYPRCTL)
+    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:
+    if not r:
-        print(f"FAILED {resp}")
+        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")

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"

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")

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:

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)

pyprland/plugins/lost_windows.py

@@ -1,9 +1,12 @@
-from .interface import Plugin
+" Moves unreachable client windows to the currently focused workspace"
+from typing import Any, cast
 
-from ..ipc import hyprctlJSON, hyprctl
+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"]
@@ -17,26 +20,26 @@ def contains(monitor, window):
     return True
 
 
-class Extension(Plugin):
+class Extension(Plugin):  # pylint: disable=missing-class-docstring
-    async def run_attract_lost(self, *args):
+    async def run_attract_lost(self):
         """Brings lost floating windows to the current workspace"""
-        monitors = await hyprctlJSON("monitors")
+        monitors = cast(list, await hyprctlJSON("monitors"))
-        windows = await hyprctlJSON("clients")
+        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 = [mon for mon in monitors if mon["focused"]][0]
+        focused: dict[str, Any] = [mon for mon in monitors if mon["focused"]][0]
         interval = focused["width"] / (1 + len(lost))
-        intervalY = focused["height"] / (1 + len(lost))
+        interval_y = focused["height"] / (1 + len(lost))
         batch = []
         workspace: int = focused["activeWorkspace"]["id"]
         margin = interval // 2
-        marginY = intervalY // 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(
+            batch.append(f'movewindowpixel exact {pos_x} {pos_y},pid:{window["pid"]}')
-                f'movewindowpixel exact {int(margin + focused["x"] + i*interval)} {int(marginY + focused["y"] + i*intervalY)},pid:{window["pid"]}'
-            )
         await hyprctl(batch)

pyprland/plugins/magnify.py

@@ -1,11 +1,10 @@
+" Toggles workspace zooming "
+from ..ipc import hyprctl
 from .interface import Plugin
 
-from ..ipc import hyprctlJSON, hyprctl
 
-
+class Extension(Plugin):  # pylint: disable=missing-class-docstring
-class Extension(Plugin):
+    zoomed = False
-    async def init(self):
-        self.zoomed = False
 
     async def run_zoom(self, *args):
         """[factor] zooms to "factor" or toggles zoom level ommited"""

pyprland/plugins/monitors.py

@@ -1,16 +1,18 @@
-from typing import Any
+" The monitors plugin "
-from .interface import Plugin
 import subprocess
+from typing import Any, cast
 
 from ..ipc import hyprctlJSON
+from .interface import Plugin
 
 
-def configure_monitors(monitors, screenid: str, x: int, y: int) -> None:
+def configure_monitors(monitors, screenid: str, pos_x: int, pos_y: int) -> None:
-    x_offset = -x if x < 0 else 0
+    "Apply the configuration change"
-    y_offset = -y if y < 0 else 0
+    x_offset = -pos_x if pos_x < 0 else 0
+    y_offset = -pos_y if pos_y < 0 else 0
 
-    min_x = x
+    min_x = pos_x
-    min_y = y
+    min_y = pos_y
 
     command = ["wlr-randr"]
     other_monitors = [mon for mon in monitors if mon["name"] != screenid]
@@ -29,61 +31,81 @@ def configure_monitors(monitors, screenid: str, x: int, y: int) -> None:
             ]
         )
 
-    command.extend(["--output", screenid, "--pos", f"{x+x_offset},{y+y_offset}"])
+    command.extend(
+        ["--output", screenid, "--pos", f"{pos_x+x_offset},{pos_y+y_offset}"]
+    )
     subprocess.call(command)
 
 
-class Extension(Plugin):
+class Extension(Plugin):  # pylint: disable=missing-class-docstring
     async def load_config(self, config) -> None:
         await super().load_config(config)
-        monitors = await hyprctlJSON("monitors")
+        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"], noDefault=True, monitors=monitors
+                monitor["name"], no_default=True, monitors=monitors
             )
 
     async def event_monitoradded(
-        self, screenid, noDefault=False, monitors: list | None = None
+        self, monitor_name, no_default=False, monitors: list | None = None
     ) -> None:
-        screenid = screenid.strip()
+        "Triggers when a monitor is plugged"
+        monitor_name = monitor_name.strip()
 
         if not monitors:
-            monitors: list[dict[str, Any]] = await hyprctlJSON("monitors")
+            monitors = cast(list, await hyprctlJSON("monitors"))
+
+        assert monitors
 
         for mon in monitors:
-            if mon["name"].startswith(screenid):
+            if mon["name"].startswith(monitor_name):
-                mon_name = mon["description"]
+                mon_description = mon["description"]
                 break
         else:
-            print(f"Monitor {screenid} not found")
+            self.log.info("Monitor %s not found", monitor_name)
             return
 
-        mon_by_name = {m["name"]: m for m in monitors}
+        if self._place_monitors(monitor_name, mon_description, monitors):
+            return
 
-        newmon = mon_by_name[screenid]
+        if not no_default:
-
-        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"]
-
-                        configure_monitors(monitors, screenid, x, y)
-                        return
-        if not noDefault:
             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[monitor_name]
+        for mon_pattern, conf in self.config["placement"].items():
+            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 = ref["x"]
+                            y = ref["y"] - newmon["height"]
+                        elif place == "bottomof":
+                            x = ref["x"]
+                            y = ref["y"] + ref["height"]
+                        elif place == "leftof":
+                            x = ref["x"] - newmon["width"]
+                            y = ref["y"]
+                        else:  # rightof
+                            x = ref["x"] + ref["width"]
+                            y = ref["y"]
+
+                        configure_monitors(monitors, monitor_name, x, y)
+                        return True
+        return False

pyprland/plugins/scratchpads.py

@@ -1,19 +1,20 @@
-import subprocess
+" Scratchpads addon "
-from typing import Any
 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_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("address") == addr:
@@ -21,22 +22,29 @@ async def get_client_props_by_address(addr: str):
 
 
 class Animations:
-    @classmethod
+    "Animation store"
-    async def fromtop(cls, monitor, client, client_uid, margin):
+
+    @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
+    @staticmethod
-    async def frombottom(cls, monitor, client, client_uid, margin):
+    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_width = int(monitor["width"] / scale)
-        mon_height = monitor["height"]
+        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
+    @staticmethod
-    async def fromleft(cls, monitor, client, client_uid, margin):
+    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
+    @staticmethod
-    async def fromright(cls, monitor, client, client_uid, margin):
+    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_width = int(monitor["width"] / scale)
-        mon_height = monitor["height"]
+        mon_height = int(monitor["height"] / scale)
 
         client_width = client["size"][0]
         client_height = client["size"][1]
@@ -72,55 +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():
+            with open(os.path.join(path, "status"), "r", encoding="utf-8") as f:
-                if line.startswith("State"):
+                for line in f.readlines():
-                    state = line.split()[1]
+                    if line.startswith("State"):
-                    return state in "RSDTt"  # not "Z (zombie)"or "X (dead)"
+                        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:
+    async def updateClientInfo(self, client_info=None) -> None:
-        if clientInfo is None:
+        "update the internal client info property, if not provided, refresh based on the current address"
-            clientInfo = await get_client_props_by_address("0x" + self.address)
+        if client_info is None:
-        assert isinstance(clientInfo, dict)
+            client_info = await get_client_props_by_address("0x" + self.address)
-        self.clientInfo.update(clientInfo)
+        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):
+class Extension(Plugin):  # pylint: disable=missing-class-docstring
-    async def init(self) -> None:
+    procs: dict[str, subprocess.Popen] = {}
-        self.procs: dict[str, subprocess.Popen] = {}
+    scratches: dict[str, Scratch] = {}
-        self.scratches: dict[str, Scratch] = {}
+    transitioning_scratches: set[str] = set()
-        self.transitioning_scratches: set[str] = set()
+    _new_scratches: set[str] = set()
-        self._respawned_scratches: set[str] = set()
+    _respawned_scratches: set[str] = set()
-        self.scratches_by_address: dict[str, Scratch] = {}
+    scratches_by_address: dict[str, Scratch] = {}
-        self.scratches_by_pid: dict[int, Scratch] = {}
+    scratches_by_pid: dict[int, Scratch] = {}
-        self.focused_window_tracking = dict()
+    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)
@@ -132,9 +164,10 @@ class Extension(Plugin):
             *(die_in_piece(scratch) for scratch in self.scratches.values())
         )
 
-    async def load_config(self, config) -> None:
+    async def load_config(self, config: dict[str, Any]) -> None:
-        config: dict[str, dict[str, Any]] = config["scratchpads"]
+        "config loader"
-        scratches = {k: Scratch(k, v) for k, v in config.items()}
+        my_config: dict[str, dict[str, Any]] = config["scratchpads"]
+        scratches = {k: Scratch(k, v) for k, v in my_config.items()}
 
         new_scratches = set()
 
@@ -147,53 +180,89 @@ 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
+        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
                     ):
+                        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)
+        "open windows hook"
-        if wrkspc.startswith("special"):
+        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:
@@ -201,65 +270,72 @@ class Extension(Plugin):
         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:
+    async def _anim_hide(self, animation_type, scratch):
-        if scratch is None:
+        "animate hiding a scratchpad"
-            for client in await hyprctlJSON("clients"):
+        addr = "address:0x" + scratch.address
-                assert isinstance(client, dict)
+        offset = scratch.conf.get("offset")
-                scratch = self.scratches_by_address.get(client["address"][2:])
+        if offset is None:
-                if not scratch:
+            if "size" not in scratch.client_info:
-                    scratch = self.scratches_by_pid.get(client["pid"])
+                await self.updateScratchInfo(scratch)
-                    if scratch:
+
-                        self.scratches_by_address[client["address"][2:]] = 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[client["address"][2:]] = scratch
-        else:
+            if scratch:
-            add_to_address_book = ("address" not in scratch.clientInfo) or (
+                await scratch.updateClientInfo(client)
-                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, autohide=False) -> None:
         """<name> hides scratchpad "name" """
         uid = uid.strip()
-        item = self.scratches.get(uid)
+        scratch = self.scratches.get(uid)
-        if not item:
+        if not scratch:
-            print(f"{uid} is not configured")
+            self.log.warning("%s is not configured", uid)
             return
-        if not item.visible and not force:
+        if not scratch.visible and not force:
-            print(f"{uid} is already hidden")
+            self.log.warning("%s is already hidden", uid)
             return
-        item.visible = False
+        scratch.visible = False
-        addr = "address:0x" + item.address
+        if not scratch.isAlive():
-        animation_type: str = item.conf.get("animation", "").lower()
+            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")
+            await self._anim_hide(animation_type, scratch)
-            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},{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 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_{uid},{addr}")
@@ -273,31 +349,46 @@ class Extension(Plugin):
                 )
                 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:
         """<name> shows scratchpad "name" """
         uid = uid.strip()
         item = self.scratches.get(uid)
 
-        self.focused_window_tracking[uid] = await hyprctlJSON("activewindow")
+        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():
+        self.log.info("Showing %s", uid)
-            print(f"{uid} is not running, restarting...")
+        await self.ensure_alive(uid, item)
-            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.start_scratch_command(uid)
-            while uid in self._respawned_scratches:
-                await asyncio.sleep(0.05)
 
         item.visible = True
         monitor = await get_focused_monitor_props()
@@ -305,6 +396,8 @@ class Extension(Plugin):
 
         await self.updateScratchInfo(item)
 
+        assert item.address, "No address !"
+
         addr = "address:0x" + item.address
 
         animation_type = item.conf.get("animation", "").lower()
@@ -317,8 +410,50 @@ class Extension(Plugin):
         if animation_type:
             margin = item.conf.get("margin", DEFAULT_MARGIN)
             fn = getattr(Animations, animation_type)
-            await fn(monitor, item.clientInfo, addr, 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 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

pyprland/plugins/shift_monitors.py

@@ -1,11 +1,17 @@
+" shift workspaces across monitors "
+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
+    monitors: list[str] = []
 
-class Extension(Plugin):
     async def init(self):
-        self.monitors = [mon["name"] for mon in await hyprctlJSON("monitors")]
+        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"""
@@ -13,13 +19,15 @@ class Extension(Plugin):
         if direction > 0:
             mon_list = self.monitors[:-1]
         else:
-            mon_list = reversed(self.monitors[1:])
+            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())

pyprland/plugins/toggle_dpms.py

@@ -1,14 +1,16 @@
+" Toggle monitors on or off "
+from typing import Any, cast
+
+from ..ipc import hyprctl, hyprctlJSON
 from .interface import Plugin
 
-from ..ipc import hyprctlJSON, hyprctl
 
-
+class Extension(Plugin):  # pylint: disable=missing-class-docstring
-class Extension(Plugin):
     async def run_toggle_dpms(self):
         """toggles dpms on/off for every monitor"""
-        monitors = await hyprctlJSON("monitors")
+        monitors = cast(list[dict[str, Any]], await hyprctlJSON("monitors"))
-        poweredOff = any(m["dpmsStatus"] for m in monitors)
+        powered_off = any(m["dpmsStatus"] for m in monitors)
-        if not poweredOff:
+        if not powered_off:
             await hyprctl("dpms on")
         else:
             await hyprctl("dpms off")

pyprland/plugins/workspaces_follow_focus.py

@@ -1,24 +1,33 @@
-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) + 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]] = []
         for n in workspaces:
@@ -36,6 +45,9 @@ 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"]

pyproject.toml

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