improve error handling & logging

README.md

@@ -11,6 +11,7 @@ Check the [wiki](https://github.com/hyprland-community/pyprland/wiki) for more i
 - Add `expose` addon
 - scratchpad: add "lazy" option
 - fix `scratchpads`'s position on monitors using scaling
+- improve error handling & logging, enable debug logs with `--debug <filename>`
 
 ## 1.3.1
 

pyprland/command.py

@@ -5,11 +5,10 @@ import sys
 import os
 import importlib
 import itertools
-import traceback
 
 
-from .ipc import get_event_stream
+from .ipc import get_event_stream, init as ipc_init
-from .common import DEBUG
+from .common import init_logger, get_logger, PyprError
 from .plugins.interface import Plugin
 
 CONTROL = f'/tmp/hypr/{ os.environ["HYPRLAND_INSTANCE_SIGNATURE"] }/.pyprland.sock'
@@ -25,11 +24,19 @@ class Pyprland:
 
     def __init__(self):
         self.plugins: dict[str, Plugin] = {}
+        self.log = get_logger()
 
     async def load_config(self, init=True):
+        try:
             self.config = json.loads(
                 open(os.path.expanduser(CONFIG_FILE), encoding="utf-8").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()
+
         for name in self.config["pyprland"]["plugins"]:
             if name not in self.plugins:
                 modname = name if "." in name else f"pyprland.plugins.{name}"
@@ -39,11 +46,16 @@ class Pyprland:
                         await plug.init()
                     self.plugins[name] = plug
                 except Exception as e:
-                    print(f"Error loading plugin {name}: {e}")
+                    self.log.error(f"Error loading plugin {name}:", exc_info=True)
-                    if DEBUG:
+                    raise PyprError()
-                        traceback.print_exc()
             if init:
+                try:
                     await self.plugins[name].load_config(self.config)
+                except PyprError:
+                    raise
+                except Exception as e:
+                    self.log.error(f"Error initializing plugin {name}:", exc_info=True)
+                    raise PyprError()
 
     async def _callHandler(self, full_name, *params):
         for plugin in [self] + list(self.plugins.values()):
@@ -51,26 +63,25 @@ class Pyprland:
                 try:
                     await getattr(plugin, full_name)(*params)
                 except Exception as e:
-                    print(f"{plugin.name}::{full_name}({params}) failed:")
+                    self.log.warn(f"{plugin.name}::{full_name}({params}) failed:")
-                    traceback.print_exc()
+                    self.log.exception(e)
 
     async def read_events_loop(self):
         while not self.stopped:
             data = (await self.event_reader.readline()).decode()
             if not data:
-                print("Reader starved")
+                self.log.critical("Reader starved")
                 return
             cmd, params = data.split(">>")
             full_name = f"event_{cmd}"
 
-            if DEBUG:
+            self.log.debug(f"EVT {full_name}({params.strip()})")
-                print(f"EVT {full_name}({params.strip()})")
             await self._callHandler(full_name, params)
 
     async def read_command(self, reader, writer) -> None:
         data = (await reader.readline()).decode()
         if not data:
-            print("Server starved")
+            self.log.critical("Server starved")
             return
         if data == "exit\n":
             self.stopped = True
@@ -91,8 +102,7 @@ class Pyprland:
         # run mako for notifications & uncomment this
         # os.system(f"notify-send '{data}'")
 
-        if DEBUG:
+        self.log.debug(f"CMD: {full_name}({args})")
-            print(f"CMD: {full_name}({args})")
 
         await self._callHandler(full_name, *args)
 
@@ -116,22 +126,31 @@ async def run_daemon():
     manager = Pyprland()
     err_count = itertools.count()
     manager.server = await asyncio.start_unix_server(manager.read_command, CONTROL)
+    max_retry = 10
+    while True:
+        attempt = next(err_count)
         try:
             events_reader, events_writer = await get_event_stream()
         except Exception as e:
-        print("Failed to get event stream: %s" % e)
+            if attempt > max_retry:
-        if next(err_count) > 10:
+                manager.log.critical(f"Failed to open hyprland event stream: {e}.")
-            raise
+                raise PyprError()
+            else:
+                manager.log.warn(
+                    f"Failed to get event stream: {e}, retry {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:
-        print(
+        raise SystemExit(1)
-            f"No config file found, create one at {CONFIG_FILE} with a valid pyprland.plugins list"
+    except Exception as e:
-        )
+        manager.log.critical(f"Failed to load config.")
         raise SystemExit(1)
 
     try:
@@ -139,7 +158,7 @@ async def run_daemon():
     except KeyboardInterrupt:
         print("Interrupted")
     except asyncio.CancelledError:
-        print("Bye!")
+        manager.log.critical("cancelled")
     finally:
         events_writer.close()
         await events_writer.wait_closed()
@@ -148,8 +167,8 @@ async def run_daemon():
 
 
 async def run_client():
-    if sys.argv[1] in ("--help", "-h"):
     manager = Pyprland()
+    if sys.argv[1] in ("--help", "-h", "help"):
         await manager.load_config(init=False)
         print(
             """Syntax: pypr [command]
@@ -171,7 +190,12 @@ Commands:
 
         return
 
+    try:
         _, writer = await asyncio.open_unix_connection(CONTROL)
+    except FileNotFoundError:
+        manager.log.critical("Failed to open control socket, is pypr daemon running ?")
+        raise PyprError()
+    else:
         writer.write((" ".join(sys.argv[1:])).encode())
         await writer.drain()
         writer.close()
@@ -179,10 +203,24 @@ Commands:
 
 
 def main():
+    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 as e:
+        log.critical(f"Command failed.")
+    except json.decoder.JSONDecodeError as e:
+        log.critical(f"Invalid JSON syntax in the config file: {e.args[0]}")
+    except Exception as e:
+        log.critical("Unhandled exception:", exc_info=True)
 
 
 if __name__ == "__main__":

pyprland/common.py

@@ -1,3 +1,64 @@
 import os
+import logging
+
+__all__ = ["DEBUG", "get_logger", "init_logger"]
 
 DEBUG = os.environ.get("DEBUG", False)
+
+
+class PyprError(Exception):
+    pass
+
+
+class LogObjects:
+    handlers: list[logging.Handler] = []
+
+
+def init_logger(filename=None, force_debug=False):
+    global DEBUG
+    if force_debug:
+        DEBUG = True
+
+    class ScreenLogFormatter(logging.Formatter):
+        LOG_FORMAT = (
+            r"%(levelname)s:%(name)s - %(message)s // %(filename)s:%(lineno)d"
+            if DEBUG
+            else r"%(levelname)s: %(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:
+        handler = logging.FileHandler(filename)
+        handler.setFormatter(
+            logging.Formatter(
+                fmt=r"%(asctime)s [%(levelname)s] %(name)s :: %(message)s :: %(filename)s:%(lineno)d"
+            )
+        )
+        LogObjects.handlers.append(handler)
+    handler = logging.StreamHandler()
+    handler.setFormatter(ScreenLogFormatter())
+    LogObjects.handlers.append(handler)
+
+
+def get_logger(name="pypr", level=None):
+    logger = logging.getLogger(name)
+    if level is None:
+        logger.setLevel(logging.DEBUG if DEBUG else logging.INFO)
+    else:
+        logger.setLevel(level)
+    logger.propagate = False
+    for handler in LogObjects.handlers:
+        logger.addHandler(handler)
+    return logger

pyprland/ipc.py

@@ -1,11 +1,13 @@
 #!/bin/env python
 import asyncio
+from logging import Logger
 from typing import Any
 import json
 import os
 
-from .common import DEBUG
+from .common import get_logger, PyprError
 
+log: Logger = None
 
 HYPRCTL = f'/tmp/hypr/{ os.environ["HYPRLAND_INSTANCE_SIGNATURE"] }/.socket.sock'
 EVENTS = f'/tmp/hypr/{ os.environ["HYPRLAND_INSTANCE_SIGNATURE"] }/.socket2.sock'
@@ -17,9 +19,12 @@ async def get_event_stream():
 
 async def hyprctlJSON(command) -> list[dict[str, Any]] | dict[str, Any]:
     """Run an IPC command and return the JSON output."""
-    if DEBUG:
+    log.debug(f"JS>> {command}")
-        print("(JS)>>>", command)
+    try:
         ctl_reader, ctl_writer = await asyncio.open_unix_connection(HYPRCTL)
+    except FileNotFoundError:
+        log.critical("hyprctl socket not found! is it running ?")
+        raise PyprError()
     ctl_writer.write(f"-j/{command}".encode())
     await ctl_writer.drain()
     resp = await ctl_reader.read()
@@ -40,9 +45,13 @@ 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:
+    log.debug(f"JS>> {command}")
-        print(">>>", command)
+    try:
         ctl_reader, ctl_writer = await asyncio.open_unix_connection(HYPRCTL)
+    except FileNotFoundError:
+        log.critical("hyprctl socket not found! is it running ?")
+        raise PyprError()
+
     if isinstance(command, list):
         ctl_writer.write(
             f"[[BATCH]] {' ; '.join(_format_command(command, base_command))}".encode()
@@ -53,11 +62,10 @@ async def hyprctl(command, base_command="dispatch") -> bool:
     resp = await ctl_reader.read(100)
     ctl_writer.close()
     await ctl_writer.wait_closed()
-    if DEBUG:
+    log.debug(f"<<JS {resp}")
-        print("<<<", resp)
     r: bool = resp == b"ok" * (len(resp) // 2)
-    if DEBUG and not r:
+    if not r:
-        print(f"FAILED {resp}")
+        log.error(f"FAILED {resp}")
     return r
 
 
@@ -67,3 +75,8 @@ async def get_focused_monitor_props() -> dict[str, Any]:
         if monitor.get("focused") == True:
             return monitor
     raise RuntimeError("no focused monitor")
+
+
+def init():
+    global log
+    log = get_logger("ipc")

pyprland/plugins/interface.py

@@ -1,9 +1,11 @@
 from typing import Any
+from ..common import get_logger
 
 
 class Plugin:
     def __init__(self, name: str):
         self.name = name
+        self.log = get_logger(name)
 
     async def init(self):
         pass

pyprland/plugins/monitors.py

@@ -55,7 +55,7 @@ class Extension(Plugin):
                 mon_name = mon["description"]
                 break
         else:
-            print(f"Monitor {screenid} not found")
+            self.log.info(f"Monitor {screenid} not found")
             return
 
         mon_by_name = {m["name"]: m for m in monitors}

pyprland/plugins/scratchpads.py

@@ -207,7 +207,7 @@ class Extension(Plugin):
         uid = uid.strip()
         item = self.scratches.get(uid)
         if not item:
-            print(f"{uid} is not configured")
+            self.log.warn(f"{uid} is not configured")
             return
         if item.visible:
             await self.run_hide(uid)
@@ -238,10 +238,10 @@ class Extension(Plugin):
         uid = uid.strip()
         item = self.scratches.get(uid)
         if not item:
-            print(f"{uid} is not configured")
+            self.log.warn(f"{uid} is not configured")
             return
         if not item.visible and not force:
-            print(f"{uid} is already hidden")
+            self.log.warn(f"{uid} is already hidden")
             return
         item.visible = False
         addr = "address:0x" + item.address
@@ -287,15 +287,15 @@ class Extension(Plugin):
         self.focused_window_tracking[uid] = await hyprctlJSON("activewindow")
 
         if not item:
-            print(f"{uid} is not configured")
+            self.log.warn(f"{uid} is not configured")
             return
 
         if item.visible and not force:
-            print(f"{uid} is already visible")
+            self.log.warn(f"{uid} is already visible")
             return
 
         if not item.isAlive():
-            print(f"{uid} is not running, restarting...")
+            self.log.info(f"{uid} is not running, restarting...")
             if uid in self.procs:
                 self.procs[uid].kill()
             if item.pid in self.scratches_by_pid: