Method using gtk_application_inhibit()

src/wakepy/methods/__init__.py

@@ -29,5 +29,6 @@
 from . import _testing as _testing
 from . import freedesktop as freedesktop
 from . import gnome as gnome
+from . import gtk as gtk
 from . import macos as macos
 from . import windows as windows

src/wakepy/methods/gtk/__init__.py

@@ -0,0 +1,6 @@
+from .gtk_application_inhibit import (
+    GtkApplicationInhibitNoIdle as GtkApplicationInhibitNoIdle,
+)
+from .gtk_application_inhibit import (
+    GtkApplicationInhibitNoSuspend as GtkApplicationInhibitNoSuspend,
+)

src/wakepy/methods/gtk/gtk_application_inhibit.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import enum
+import typing
+from abc import ABC, abstractmethod
+
+from wakepy.core import Method, ModeName, PlatformType
+from wakepy.pyinhibitor import get_inhibitor
+
+if typing.TYPE_CHECKING:
+    from typing import Optional
+
+
+class GtkIhibitFlag(enum.IntFlag):
+    """The ApplicationInhibitFlags from
+    https://docs.gtk.org/gtk4/flags.ApplicationInhibitFlags.html
+    """
+
+    # Inhibit suspending the session or computer
+    INHIBIT_SUSPEND = 4
+    # Inhibit the session being marked as idle (and possibly locked).
+    INHIBIT_IDLE = 8
+
+
+class _GtkApplicationInhibit(Method, ABC):
+    """Method using the gtk_application_inhibit().
+
+    https://docs.gtk.org/gtk4/method.Application.inhibit.html
+
+    Works on GTK 3 and 4.
+    """
+
+    supported_platforms = (PlatformType.UNIX_LIKE_FOSS,)
+    inhibitor_module = "wakepy.methods.gtk.inhibitor"
+
+    @property
+    @abstractmethod
+    def flags(self) -> GtkIhibitFlag: ...
+
+    def __init__(self, **kwargs: object) -> None:
+        super().__init__(**kwargs)
+        self.inhibit_cookie: Optional[int] = None
+
+    def enter_mode(self) -> None:
+        self.inhibitor = get_inhibitor(self.inhibitor_module)
+        # TODO: use flags
+        self.inhibitor.start()
+
+    def exit_mode(self) -> None:
+        self.inhibitor.stop()
+
+
+class GtkApplicationInhibitNoSuspend(_GtkApplicationInhibit):
+    name = "gtk_application_inhibit"
+    mode_name = ModeName.KEEP_RUNNING
+    flags = GtkIhibitFlag.INHIBIT_SUSPEND
+
+
+class GtkApplicationInhibitNoIdle(_GtkApplicationInhibit):
+    name = "gtk_application_inhibit"
+    mode_name = ModeName.KEEP_PRESENTING
+    flags = GtkIhibitFlag.INHIBIT_IDLE | GtkIhibitFlag.INHIBIT_SUSPEND

src/wakepy/methods/gtk/inhibitor.py

@@ -0,0 +1,99 @@
+"""Inhibitor module which uses gtk_application_inhibit(), and follows the
+inhibitor module specification (can be used with the pyinhibitor server).
+
+NOTE: Due to the inhibitor counter, this module should not be executed (via a
+forced re-import) more than once."""
+
+from __future__ import annotations
+
+import logging
+import threading
+import warnings
+
+with warnings.catch_warnings():
+    # Ignore the PyGIWarning: Gtk was imported without specifying a version
+    # first. This should work on GtK 3 and 4.
+    warnings.filterwarnings(action="ignore")
+    from gi.repository import Gio, Gtk
+
+logger = logging.getLogger(__name__)
+latest_inhibitor_identifier = 0
+
+lock = threading.Lock()
+
+
+class Inhibitor:
+    """Inhibitor which uses GTK, namely the gtk_application_inhibit()
+
+    Docs: https://docs.gtk.org/gtk3/method.Application.inhibit.html
+    """
+
+    def __init__(self):
+        self.app: Gtk.Application | None = None
+        self.cookie: int | None = None
+
+    def start(self, *_) -> None:
+        self.app = self._get_app()
+        try:
+
+            # Docs: https://lazka.github.io/pgi-docs/#Gtk-4.0/classes/Application.html#Gtk.Application.inhibit
+            cookie = self.app.inhibit(
+                Gtk.ApplicationWindow(application=self.app),
+                Gtk.ApplicationInhibitFlags(8),  # prevent idle
+                "wakelock requested (wakepy)",
+            )
+            if not cookie:
+                raise RuntimeError(
+                    "Failed to inhibit the system (Gtk.Application.inhibit did not "
+                    "return a non-zero cookie)"
+                )
+
+            self.cookie = cookie
+
+            # The hold() keeps the app alive even without a window.
+            # Docs: https://lazka.github.io/pgi-docs/Gio-2.0/classes/Application.html#Gio.Application.hold
+            self.app.hold()
+
+        except Exception as error:
+            self.app.quit()
+            raise RuntimeError(f"Failed to inhibit the system: {error}")
+
+    def _get_app(self) -> Gtk.Application:
+        lock.acquire()
+        # NOTE: Cannot register two apps with same applidation_id within the
+        # same python process (not even on separate threads)! In addition,
+        # quitting the app does not seem to unregister / make it possible to
+        # reuse the application_id. Therefore, using unique application_id for
+        # each instance.
+        global latest_inhibitor_identifier
+        latest_inhibitor_identifier += 1
+        application_id = f"io.readthedocs.wakepy.inhibitor{latest_inhibitor_identifier}"
+        try:
+            app = Gtk.Application(
+                application_id=application_id,
+                flags=Gio.ApplicationFlags.IS_SERVICE | Gio.ApplicationFlags.NON_UNIQUE,
+            )
+
+            # Cannot use the inhibit() if the app is not registered first.
+
+            logger.debug("Registering Gtk.Application with id  %s", application_id)
+            app.register()
+            logger.debug("Registered Gtk.Application with id  %s", application_id)
+        except Exception as error:
+            raise RuntimeError(
+                f"Failed to create or register the Gtk.Application: {error}"
+            ) from error
+        finally:
+            lock.release()
+
+        return app
+
+    def stop(self) -> None:
+        if self.cookie:
+            # Counterpart to hold().
+            self.app.uninhibit(self.cookie)
+            self.cookie = None
+
+        self.app.release()
+        self.app.quit()
+        self.app = None

src/wakepy/pyinhibitor/__init__.py

@@ -0,0 +1,15 @@
+"""This subpackage defines the python based inhibitor client and server.
+
+With this package, it is possible to use python packages outside of your
+current python environment (for example, directly from your system python
+site-packages). The idea is to run a python server with the required packages
+and communicate with it via a unix socket.
+
+The server can only utilize "inhibitor modules", which are simple python
+modules that define a class called Inhibitor. See the inhibitor_server.py
+for the full specification.
+
+This works only on unix-like systems (on systems which support unix sockets).
+"""
+
+from .inhibitors import get_inhibitor as get_inhibitor

src/wakepy/pyinhibitor/inhibitor_server.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+import importlib.util
+import sys
+import typing
+import warnings
+from pathlib import Path
+from socket import AF_UNIX, SOCK_STREAM, socket
+
+if sys.version_info < (3, 8):  # pragma: no-cover-if-py-gte-38
+    from typing_extensions import Protocol
+else:  # pragma: no-cover-if-py-lt-38
+    from typing import Protocol
+
+if typing.TYPE_CHECKING:
+    from typing import Type
+
+
+class Inhibitor(Protocol):
+    """The Inhibitor protocol. An inhibitor module should provide a class
+    called Inhibitor which implements this protocol."""
+
+    def start(self, *args) -> None: ...
+    def stop(self) -> None: ...
+
+
+CLIENT_CONNECTION_TIMEOUT = 60
+"""Time to wait (seconds) for the client to connect to the server."""
+CLIENT_MESSAGE_TIMEOUT = 1
+"""Time to wait (seconds) for each message from the client."""
+
+
+class InhibitorServer:
+    """A very simple class for inhibiting suspend/idle.
+
+    Communicates with a main process using a Unix domain socket.
+
+    What happens when run() is called:
+    1. When the process starts, inhibit() is called. If it succeeds, this
+       process sends "INHIBIT_OK". If it fails, this process sends
+       "INHIBIT_ERROR" and exits.
+    2. This process waits indefinitely for a "QUIT" message.
+    3. When "QUIT" (or empty string) is received, uninhibit() is called. If it
+        succeeds, this process sends "UNINHIBIT_OK". If it fails, this process
+        sends "UNINHIBIT_ERROR". Then, this process exits.
+    """
+
+    def __init__(self):
+        self._inhibitor: Inhibitor | None = None
+
+    def run(self, socket_path: str, inhibitor_module: str, *inhibit_args) -> None:
+        """Inhibit the system using inhibitor_module and wait for a quit
+        message at socket_path.
+
+        Parameters
+        ----------
+        inhibitor_module : str
+            The python module that contains the Inhibitor class
+        socket_path : str
+            The path to the Unix domain socket which is used for communication.
+        """
+        server_socket = socket(AF_UNIX, SOCK_STREAM)
+        Path(socket_path).expanduser().unlink(missing_ok=True)
+        server_socket.bind(socket_path)
+
+        try:
+            self._run(server_socket, inhibitor_module, *inhibit_args)
+        finally:
+            server_socket.close()
+
+    def _run(self, server_socket: socket, inhibitor_module: str, *inhibit_args) -> None:
+        server_socket.listen(1)  # Only allow 1 connection at a time
+        client_socket = self._get_client_socket(server_socket)
+        client_socket.settimeout(CLIENT_MESSAGE_TIMEOUT)
+
+        try:
+            self.inhibit(inhibitor_module, *inhibit_args)
+            self.send_message(client_socket, "INHIBIT_OK")
+        except Exception as error:
+            self.send_message(client_socket, f"INHIBIT_ERROR:{error}")
+            sys.exit(0)
+
+        while True:
+            # Called every `CLIENT_MESSAGE_TIMEOUT` seconds.
+            should_quit = self.check_for_quit_message(client_socket)
+            if should_quit:
+                break
+
+        try:
+            self.uninhibit()
+            self.send_message(client_socket, "UNINHIBIT_OK")
+        except Exception as error:
+            self.send_message(client_socket, f"UNINHIBIT_ERROR:{error}")
+            sys.exit(0)
+
+    @staticmethod
+    def _get_client_socket(server_socket: socket) -> socket:
+        server_socket.settimeout(CLIENT_CONNECTION_TIMEOUT)
+
+        try:
+            client_socket, _ = server_socket.accept()
+        except TimeoutError as e:
+            raise TimeoutError(
+                f"Client did not connect within {CLIENT_CONNECTION_TIMEOUT} seconds."
+            ) from e
+        except KeyboardInterrupt:
+            print("Interrupted manually. Exiting.")
+            sys.exit(0)
+
+        return client_socket
+
+    def inhibit(self, inhibitor_module: str, *inhibit_args) -> None:
+        """Inhibit using the Inhibitor class in the given `inhibitor_module`.
+        In case the operation fails, raises a RuntimeError."""
+        inhibitor_class = self.get_inhibitor_class(inhibitor_module)
+        self._inhibitor = inhibitor_class()
+        self._inhibitor.start(*inhibit_args)
+
+    @staticmethod
+    def get_inhibitor_class(inhibitor_module_path: str) -> Type[Inhibitor]:
+        try:
+            module_name = "__wakepy_inhibitor"
+            spec = importlib.util.spec_from_file_location(
+                module_name, inhibitor_module_path
+            )
+            module = importlib.util.module_from_spec(spec)
+            sys.modules[module_name] = module
+            spec.loader.exec_module(module)
+        except ImportError as e:
+            raise ImportError(
+                f"{e} | Used python interpreter: {sys.executable}."
+            ) from e
+        return module.Inhibitor
+
+    def uninhibit(self) -> None:
+        """Uninhibit what was inhibited. In case the operation fails, raises a
+        RuntimeError."""
+        if self._inhibitor:
+            self._inhibitor.stop()
+            self._inhibitor = None
+        else:
+            warnings.warn("Called uninhibit before inhibit -> doing nothing.")
+
+    def send_message(self, client_socket: socket, message: str) -> None:
+        client_socket.sendall(message.encode())
+
+    def check_for_quit_message(self, sock: socket) -> bool:
+        # waits until the socket gets a message
+        try:
+            request = sock.recv(1024).decode()
+        except TimeoutError:
+            return False
+        print(f"Received request: {request}")
+        # if the client disconnects, empty string is returned. This will make
+        # sure that the server process quits automatically when it's not needed
+        # anymore.
+        return request == "QUIT" or request == ""
+
+
+if __name__ == "__main__":
+    if len(sys.argv) < 3:
+        print(
+            f"Usage: python {__file__} <socket_path> <inhibitor_module> "
+            "[inhibit_args...]"
+        )
+        sys.exit(1)
+
+    # Get the socket path from the command-line arguments
+    InhibitorServer().run(
+        socket_path=sys.argv[1], inhibitor_module=sys.argv[2], *sys.argv[3:]
+    )