From c954020f7f3e30966b6da1942769dc5f0a6d1a55 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Niko=20F=C3=B6hr?= <17479683+fohrloop@users.noreply.github.com> Date: Fri, 4 Jul 2025 14:40:47 +0300 Subject: [PATCH] gtk_application_inhibit based wakepy Method --- docs/source/changelog.md | 4 +- src/wakepy/methods/__init__.py | 1 + src/wakepy/methods/gtk/__init__.py | 6 + .../methods/gtk/gtk_application_inhibit.py | 62 ++++ src/wakepy/methods/gtk/inhibitor.py | 101 +++++++ src/wakepy/pyinhibitor/__init__.py | 15 + src/wakepy/pyinhibitor/inhibitor_server.py | 176 ++++++++++++ src/wakepy/pyinhibitor/inhibitors.py | 271 ++++++++++++++++++ src/wakepy/utils.py | 54 ++++ tests/unit/test_pyinhibitor.py | 22 ++ 10 files changed, 711 insertions(+), 1 deletion(-) create mode 100644 src/wakepy/methods/gtk/__init__.py create mode 100644 src/wakepy/methods/gtk/gtk_application_inhibit.py create mode 100644 src/wakepy/methods/gtk/inhibitor.py create mode 100644 src/wakepy/pyinhibitor/__init__.py create mode 100644 src/wakepy/pyinhibitor/inhibitor_server.py create mode 100644 src/wakepy/pyinhibitor/inhibitors.py create mode 100644 src/wakepy/utils.py create mode 100644 tests/unit/test_pyinhibitor.py diff --git a/docs/source/changelog.md b/docs/source/changelog.md index b0f6479b..d6d283b3 100644 --- a/docs/source/changelog.md +++ b/docs/source/changelog.md @@ -4,15 +4,17 @@ 🗓️ Unreleased ### ✨ Features +- New method: "gtk_application_inhibit", which calls the [gtk_application_inhibit()](https://docs.gtk.org/gtk4/method.Application.inhibit.html) through the GObject Instrospection python package (import name `gi`). This supports both GTK3 and GTK4 based Desktop Environments (examples: GNOME, Xfce, Cinnamon, LXDE, MATE, Unity, Budgie and Pantheon). This can use _system python_, so installing PyGObject is not needed. ([#407](https://github.com/fohrloop/wakepy/pull/407)) - Update the wakepy CLI printout: Adds the used Method and activated Mode to the printout ([#434](https://github.com/fohrloop/wakepy/pull/434)) - Added `-v` (INFO) and `-vv` (DEBUG) verbosity flags for the [wakepy CLI command](https://wakepy.readthedocs.io/stable/cli-api.html). ([#439](https://github.com/fohrloop/wakepy/pull/439)) - Add [Mode.method](https://wakepy.readthedocs.io/stable/api-reference.html#wakepy.Mode.method), [ActivationResult.method](https://wakepy.readthedocs.io/stable/api-reference.html#wakepy.ActivationResult.method) and [MethodActivationResult.method](https://wakepy.readthedocs.io/stable/api-reference.html#wakepy.MethodActivationResult.method) attributes, which are instances of [MethodInfo](https://wakepy.readthedocs.io/stable/api-reference.html#wakepy.MethodInfo) ([#459](https://github.com/fohrloop/wakepy/pull/459), [#460](https://github.com/fohrloop/wakepy/pull/460), [#464](https://github.com/fohrloop/wakepy/pull/464)) - Make [Mode.active_method](https://wakepy.readthedocs.io/stable/api-reference.html#wakepy.Mode.active_method) a [MethodInfo](https://wakepy.readthedocs.io/stable/api-reference.html#wakepy.MethodInfo) instance (was a string) ([#459](https://github.com/fohrloop/wakepy/pull/459)) -### ✨ Minor Enhancements +### ✨ Enhancements - Better error messages: When selected Method is not part of the selected Mode ([#427](https://github.com/fohrloop/wakepy/pull/427)) and when a D-Bus -based method fails ([#438](https://github.com/fohrloop/wakepy/pull/438)) - Enhance Mode Activation Observability. Add logging (DEBUG and INFO level) for different parts in the Mode activation process. Show which methods are to be tried, and log any success and failure of activating a Mode. Improved the `ActivationResult.get_failure_text()` output. Added `NoMethodsWarning` which is issued if trying to activate a Mode with an empty list of methods. ([#411](https://github.com/fohrloop/wakepy/pull/411)) - Make ActivationWarning use proper stacklevel, so that issued warnings point to user code, and not into wakepy source code. ([#432](https://github.com/fohrloop/wakepy/pull/432)) +- Better error message when selected Method is not part of the selected Mode ([#427](https://github.com/fohrloop/wakepy/pull/427)) ### 🐞 Bug fixes - Fix prioritized order of Methods ([#429](https://github.com/fohrloop/wakepy/pull/429)). diff --git a/src/wakepy/methods/__init__.py b/src/wakepy/methods/__init__.py index caeea99e..454f8e32 100644 --- a/src/wakepy/methods/__init__.py +++ b/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 diff --git a/src/wakepy/methods/gtk/__init__.py b/src/wakepy/methods/gtk/__init__.py new file mode 100644 index 00000000..e1135259 --- /dev/null +++ b/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, +) diff --git a/src/wakepy/methods/gtk/gtk_application_inhibit.py b/src/wakepy/methods/gtk/gtk_application_inhibit.py new file mode 100644 index 00000000..ff15e968 --- /dev/null +++ b/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 diff --git a/src/wakepy/methods/gtk/inhibitor.py b/src/wakepy/methods/gtk/inhibitor.py new file mode 100644 index 00000000..f28ceba4 --- /dev/null +++ b/src/wakepy/methods/gtk/inhibitor.py @@ -0,0 +1,101 @@ +"""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. + # Basically increments the internal hold count of the application. + # 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: + self.app.uninhibit(self.cookie) + self.cookie = None + + # The app.release is the counterpart to app.hold(); decrement the internal + # hold count of the application. + self.app.release() + self.app.quit() + self.app = None diff --git a/src/wakepy/pyinhibitor/__init__.py b/src/wakepy/pyinhibitor/__init__.py new file mode 100644 index 00000000..bacff84e --- /dev/null +++ b/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 diff --git a/src/wakepy/pyinhibitor/inhibitor_server.py b/src/wakepy/pyinhibitor/inhibitor_server.py new file mode 100644 index 00000000..f13bcbf8 --- /dev/null +++ b/src/wakepy/pyinhibitor/inhibitor_server.py @@ -0,0 +1,176 @@ +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:{errortext}" 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 + ---------- + socket_path : str + The path to the Unix domain socket which is used for communication. + inhibitor_module : str + The python module that contains the Inhibitor class + inhibit_args: + Any arguments to the Inhibitor.start() method. + """ + 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__": + # This is the entry point for the inhibitor server, and it's called + # automatically when using the start_inhibit_server() + + if len(sys.argv) < 3: + print( + f"Usage: python {__file__} " + "[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:] + ) diff --git a/src/wakepy/pyinhibitor/inhibitors.py b/src/wakepy/pyinhibitor/inhibitors.py new file mode 100644 index 00000000..63bc082f --- /dev/null +++ b/src/wakepy/pyinhibitor/inhibitors.py @@ -0,0 +1,271 @@ +from __future__ import annotations + +import logging +import subprocess +import sys +import tempfile +import time +import uuid +from importlib import import_module +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 + +logger = logging.getLogger(__name__) + +# Path to the Unix domain socket file +SOCKET_PATH_TEMPLATE = "{tmpdir}/wakepy/wakepy-pyinhibit-subprocess-{id}.socket" + +INHIBITOR_SERVER_MODULE_PATH = Path(__file__).parent / "inhibitor_server.py" + + +def get_system_python_path() -> str: + """Gets the path to the system python interpreter. Works only on Unix-like + systems.""" + alternatives = [ + "/usr/bin/python3", + "/usr/bin/python", + "/usr/local/bin/python3", + "/usr/local/bin/python", + ] + for path in alternatives: + if Path(path).exists(): + return str(Path(path).expanduser().resolve()) + raise RuntimeError( + f"Could not find the system python interpreter. Tried paths: {alternatives}" + ) + + +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: ... + + +def get_inhibitor(inhibitor_module: str) -> Inhibitor: + """Import the inhibitor module from path specified by `inhibitor_module` + and return the Inhibitor class. If the module is not found in the current + python environment, a SubprocessInhibor using system python is returned, + instead.""" + + try: + # If the import_module fails, this block takes like 2 milliseconds, + # which is about 1% of the time the SubprocessInhibor takes to start + # using an inhibitor which uses the 'gi' module, for example. In other + # words, it is pretty cheap operation to try to import the module first + # in the local python environment. + module = import_module(inhibitor_module) + inhibitor = module.Inhibitor() + logger.debug( + "Inhibitor module '%s' loaded to local python environment", inhibitor_module + ) + return inhibitor + except ImportError as exc: + system_python_path = get_system_python_path() + inhibitor = SubprocessInhibor( + socket_path=get_socket_path(), + python_path=system_python_path, + inhibitor_path=get_module_path(inhibitor_module), + ) + logger.debug( + 'ImportError while importing the Inhibitor module "%s":\n %s\n' + ' Trying to use "%s" instead.', + inhibitor_module, + str(exc), + system_python_path, + ) + return inhibitor + + +def get_socket_path() -> str: + socket_path = SOCKET_PATH_TEMPLATE.format( + tmpdir=tempfile.gettempdir(), id=uuid.uuid4() + ) + Path(socket_path).parent.mkdir(parents=True, exist_ok=True) + return socket_path + + +def get_module_path(inhibitor_module: str) -> Path: + """Get the path to the module specified by `inhibitor_module`. + + Parameters + ---------- + inhibitor_module : str + The module path, like "wakepy.methods.gtk.inhibitor". Note that this + function only supports modules (not packages), and that the file + extension of the modules is assumed to be ".py". All module paths must + start with "wakepy." + + Returns + ------- + Path: + The path to the module file. For example: + PosixPath('/home/user/venv/wakepy/methods/gtk/inhibitor.py') + + """ + import wakepy + + if not inhibitor_module.startswith("wakepy."): + raise ValueError("The module path must start with 'wakepy.'") + + wakepy_path = Path(wakepy.__file__).parent + path_parts = inhibitor_module.split(".")[1:] + return wakepy_path.joinpath(*path_parts).with_suffix(".py") + + +class SubprocessInhibor: + """Runs an Inhibitor in a subprocess; Runs aa inhibitor server with the given + python interpreter and the inhibitor module. This is an alternative way + of using an inhibitor module (needed when required modules are not + available in the current python environment).""" + + def __init__( + self, + socket_path: str, + python_path: str, + inhibitor_path: str, + ): + self.socket_path = socket_path + self.python_path = python_path + self.inhibitor_path = inhibitor_path + self.client_socket: socket | None = None + + def start(self, *args) -> None: + + start_inhibit_server( + self.socket_path, + self.python_path, + self.inhibitor_path, + *args, + ) + + self.client_socket = get_client_socket(self.socket_path) + + try: + get_and_handle_inhibit_result(self.client_socket) + except Exception: + self.client_socket.close() + raise + + def stop(self) -> None: + + try: + send_quit(self.client_socket) + finally: + self.client_socket.close() + + Path(self.socket_path).unlink(missing_ok=True) + + +def start_inhibit_server( + socket_path: str, python_path: str, inhibitor_path: str, *inhibitor_args: object +): + """Starts the pyinhibitor server. + + Parameters + ---------- + python_path : str + The path to the python interpreter + inhibitor_path: str + The path to the inhibitor python module. This module must contain a + class called Inhibitor which implements the Inhibitor protocol. + """ + socket_pth = Path(socket_path).expanduser() + # Remove the file so we can just wait the file to appear and know that + # the server is ready. + socket_pth.unlink(missing_ok=True) + + cmd = [ + str(x) + for x in ( + python_path, + INHIBITOR_SERVER_MODULE_PATH, + socket_path, + inhibitor_path, + *inhibitor_args, + ) + ] + logger.debug("Starting the inhibitor server with command: %s", " ".join(cmd)) + subprocess.Popen(cmd) + try: + # On an old 2018 HP Elitebook 840 G5, the server takes about 0.17 + # seconds to create the socket file. Hence, assuming that the wait time + # is never more than 2 seconds.. + wait_until_file_exists(socket_pth, total_wait_time=2) + except Exception as e: + raise RuntimeError( + f"Something went wrong while calling {cmd}. Details: '{e}'" + ) from e + + +def wait_until_file_exists( + file_path: Path, total_wait_time: float = 2, wait_time_per_cycle=0.001 +) -> None: + """Waits until a file exists or the total_wait_time is reached. + + Parameters + ---------- + file_path : Path + The path to the file + total_wait_time : float, optional + The total time to wait. Default: 2 (seconds) + wait_time_per_cycle : float, optional + The time to wait between each cycle. Default: 0.001 (seconds) + + Raises + ------ + FileNotFoundError + If the file does not exist after the total_wait_time + """ + + for _ in range(int(total_wait_time / wait_time_per_cycle)): + if file_path.exists(): + break + time.sleep(wait_time_per_cycle) + else: + raise FileNotFoundError( + f"File {file_path} does not exists. (Waited {total_wait_time} seconds)." + ) + + +def get_client_socket(socket_path: str) -> socket: + client_socket = socket(AF_UNIX, SOCK_STREAM) + try: + client_socket.connect(socket_path) + except ConnectionRefusedError: + raise RuntimeError("Must start the server first.") + client_socket.settimeout(1) + return client_socket + + +def get_and_handle_inhibit_result(client_socket: socket): + response = _get_response_from_server(client_socket) + + if response.startswith("INHIBIT_ERROR"): + errtext = response.split(":", maxsplit=1)[1] + raise RuntimeError(errtext) + elif response != "INHIBIT_OK": # should never happen + raise RuntimeError("Failed to inhibit the system") + + +def send_quit(client_socket: socket): + client_socket.sendall("QUIT".encode()) + response = _get_response_from_server(client_socket) + + if response.startswith("UNINHIBIT_ERROR"): + errtext = response.split(":", maxsplit=1)[-1] + raise RuntimeError(f"Failed to uninhibit the system: {errtext}") + elif response != "UNINHIBIT_OK": # should never happen + raise RuntimeError("Failed to uninhibit the system") + + +def _get_response_from_server(client_socket: socket) -> str: + response = client_socket.recv(1024).decode() + logger.debug("Response from pyinhibitor server: %s", response) + return response diff --git a/src/wakepy/utils.py b/src/wakepy/utils.py new file mode 100644 index 00000000..56870ae5 --- /dev/null +++ b/src/wakepy/utils.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import subprocess +import time +import typing +from pathlib import Path + +if typing.TYPE_CHECKING: + from typing import Iterable + +search_directories = [ + "/usr/bin", # default python used by a unix system + "/usr/local/bin", # python installed by the user is typically here +] + + +def get_python_path(required_modules: list[str] | None) -> str | None: + """Gets the path to the system python interpreter which has the required + modules.""" + for python in iter_python3_executable_paths(search_directories): + if required_modules: + if not has_modules(python, required_modules): + continue + return python + + +def iter_python3_executable_paths(directories: Iterable[str]): + for directory in directories: + executable = Path(directory) / "python3" + if not executable.exists(): + continue + yield executable + + +def has_modules(python: str, modules: list[str]) -> bool: + """Checks if the python interpreter has the required modules.""" + t1 = time.time() + cmd = ( + "from importlib.util import find_spec;" + f"""print(all(find_spec(module) for module in {modules}),end='')""" + ) + out = subprocess.run([python, "-c", cmd], check=True, capture_output=True) + t2 = time.time() + print("took", t2 - t1) + return out.stdout == b"True" + + +if __name__ == "__main__": + import time + + t1 = time.time() + print(get_python_path(["gi"])) + t2 = time.time() + print(t2 - t1) diff --git a/tests/unit/test_pyinhibitor.py b/tests/unit/test_pyinhibitor.py new file mode 100644 index 00000000..7c3fe82e --- /dev/null +++ b/tests/unit/test_pyinhibitor.py @@ -0,0 +1,22 @@ +from pathlib import Path + +import pytest + +import wakepy +from wakepy.pyinhibitor.inhibitors import get_module_path + + +class TestGetModulePath: + + def test_four_levels(self): + assert ( + get_module_path("wakepy.methods.gtk.inhibitor") + == Path(wakepy.__file__).parent / "methods" / "gtk" / "inhibitor.py" + ) + + def test_two_levels(self): + assert get_module_path("wakepy.foo") == Path(wakepy.__file__).parent / "foo.py" + + def test_bad_path(self): + with pytest.raises(ValueError): + get_module_path("foo.bar")