Merge and trim the two utils modules (#13)

rendercanvas/__init__.py

@@ -5,7 +5,7 @@
 # ruff: noqa: F401
 
 from ._version import __version__, version_info
-from . import _gui_utils
+from . import _coreutils
 from ._events import EventType
 from .base import BaseRenderCanvas, BaseLoop, BaseTimer
 

rendercanvas/_coreutils.py

@@ -2,36 +2,25 @@
 Core utilities that are loaded into the root namespace or used internally.
 """
 
+import os
 import re
 import sys
 import types
-import atexit
+import weakref
 import logging
-import importlib.resources
-from contextlib import ExitStack
+import ctypes.util
+from contextlib import contextmanager
 
 
-# Our resources are most probably always on the file system. But in
-# case they don't we have a nice exit handler to remove temporary files.
-_resource_files = ExitStack()
-atexit.register(_resource_files.close)
-
-
-def get_resource_filename(name):
-    """Get the filename to a wgpu resource."""
-    if sys.version_info < (3, 9):
-        context = importlib.resources.path("wgpu.resources", name)
-    else:
-        ref = importlib.resources.files("wgpu.resources") / name
-        context = importlib.resources.as_file(ref)
-    path = _resource_files.enter_context(context)
-    return str(path)
+# %% Logging
 
 
 logger = logging.getLogger("rendercanvas")
 logger.setLevel(logging.WARNING)
 
 
+err_hashes = {}
+
 _re_wgpu_ob = re.compile(r"`<[a-z|A-Z]+-\([0-9]+, [0-9]+, [a-z|A-Z]+\)>`")
 
 
@@ -42,6 +31,55 @@ def error_message_hash(message):
     return hash(message)
 
 
+@contextmanager
+def log_exception(kind):
+    """Context manager to log any exceptions, but only log a one-liner
+    for subsequent occurrences of the same error to avoid spamming by
+    repeating errors in e.g. a draw function or event callback.
+    """
+    try:
+        yield
+    except Exception as err:
+        # Store exc info for postmortem debugging
+        exc_info = list(sys.exc_info())
+        exc_info[2] = exc_info[2].tb_next  # skip *this* function
+        sys.last_type, sys.last_value, sys.last_traceback = exc_info
+        # Show traceback, or a one-line summary
+        msg = str(err)
+        msgh = error_message_hash(msg)
+        if msgh not in err_hashes:
+            # Provide the exception, so the default logger prints a stacktrace.
+            # IDE's can get the exception from the root logger for PM debugging.
+            err_hashes[msgh] = 1
+            logger.error(kind, exc_info=err)
+        else:
+            # We've seen this message before, return a one-liner instead.
+            err_hashes[msgh] = count = err_hashes[msgh] + 1
+            msg = kind + ": " + msg.split("\n")[0].strip()
+            msg = msg if len(msg) <= 70 else msg[:69] + "…"
+            logger.error(msg + f" ({count})")
+
+
+# %% Weak bindings
+
+
+def weakbind(method):
+    """Replace a bound method with a callable object that stores the `self` using a weakref."""
+    ref = weakref.ref(method.__self__)
+    class_func = method.__func__
+    del method
+
+    def proxy(*args, **kwargs):
+        self = ref()
+        if self is not None:
+            return class_func(self, *args, **kwargs)
+
+    proxy.__name__ = class_func.__name__
+    return proxy
+
+
+# %% Enum
+
 # We implement a custom enum class that's much simpler than Python's enum.Enum,
 # and simply maps to strings or ints. The enums are classes, so IDE's provide
 # autocompletion, and documenting with Sphinx is easy. That does mean we need a
@@ -112,101 +150,100 @@ def __init__(self):
         raise RuntimeError("Cannot instantiate an enum.")
 
 
-_flag_cache = {}  # str -> int
+# %% lib support
 
 
-def str_flag_to_int(flag, s):
-    """Allow using strings for flags, i.e. 'READ' instead of wgpu.MapMode.READ.
-    No worries about repeated overhead, because the results are cached.
-    """
-    cache_key = f"{flag.__name__}.{s}"  # use class name
-    value = _flag_cache.get(cache_key, None)
-
-    if value is None:
-        parts = [p.strip() for p in s.split("|")]
-        parts = [p for p in parts if p]
-        invalid_parts = [p for p in parts if p.startswith("_")]
-        if not parts or invalid_parts:
-            raise ValueError(f"Invalid flag value: {s}")
-
-        value = 0
-        for p in parts:
-            try:
-                v = flag.__dict__[p.upper()]
-                value += v
-            except KeyError:
-                raise ValueError(f"Invalid flag value for {flag}: '{p}'") from None
-        _flag_cache[cache_key] = value
-
-    return value
-
-
-class ApiDiff:
-    """Helper class to define differences in the API by annotating
-    methods. This way, these difference are made explicit, plus they're
-    logged so we can automatically include these changes in the docs.
+QT_MODULE_NAMES = ["PySide6", "PyQt6", "PySide2", "PyQt5"]
+
+
+def get_imported_qt_lib():
+    """Get the name of the currently imported qt lib.
+
+    Returns (name, has_application). The name is None when no qt lib is currently imported.
     """
 
-    def __init__(self):
-        self.hidden = {}
-        self.added = {}
-        self.changed = {}
-
-    def hide(self, func_or_text):
-        """Decorator to discard certain methods from the "reference" API.
-        Intended only for the base API where we deviate from WebGPU.
-        """
-        return self._diff("hidden", func_or_text)
-
-    def add(self, func_or_text):
-        """Decorator to add certain methods that are not part of the "reference" spec.
-        Intended for the base API where we implement additional/alternative API,
-        and in the backend implementations where additional methods are provided.
-        """
-        return self._diff("added", func_or_text)
-
-    def change(self, func_or_text):
-        """Decorator to mark certain methods as having a different signature
-        as the "reference" spec. Intended only for the base API where we deviate
-        from WebGPU.
-        """
-        return self._diff("changed", func_or_text)
-
-    def _diff(self, method, func_or_text):
-        def wrapper(f):
-            d = getattr(self, method)
-            name = f.__qualname__ if hasattr(f, "__qualname__") else f.fget.__qualname__
-            d[name] = text
-            return f
-
-        if callable(func_or_text):
-            text = None
-            return wrapper(func_or_text)
-        else:
-            text = func_or_text
-            return wrapper
-
-    def remove_hidden_methods(self, scope):
-        """Call this to remove methods from the API that were decorated as hidden."""
-        for name in self.hidden:
-            classname, _, methodname = name.partition(".")
-            cls = scope[classname]
-            delattr(cls, methodname)
-
-    @property
-    def __doc__(self):
-        """Generate a docstring for this instance. This way we can
-        automatically document API differences.
-        """
-        lines = [""]
-        for name, msg in self.hidden.items():
-            line = f"    * Hides ``{name}()``"
-            lines.append(f"{line} - {msg}" if msg else line)
-        for name, msg in self.added.items():
-            line = f"    * Adds ``{name}()``"
-            lines.append(f"{line} - {msg}" if msg else line)
-        for name, msg in self.changed.items():
-            line = f"    * Changes ``{name}()``"
-            lines.append(f"{line} - {msg}" if msg else line)
-        lines.append("")
-        return "\n".join(sorted(lines))
+    # Get all imported qt libs
+    imported_libs = []
+    for libname in QT_MODULE_NAMES:
+        qtlib = sys.modules.get(libname, None)
+        if qtlib is not None:
+            imported_libs.append(libname)
+
+    # Get which of these have an application object
+    imported_libs_with_app = []
+    for libname in imported_libs:
+        QtWidgets = sys.modules.get(libname + ".QtWidgets", None)  # noqa: N806
+        if QtWidgets:
+            app = QtWidgets.QApplication.instance()
+            if app is not None:
+                imported_libs_with_app.append(libname)
+
+    # Return findings
+    if imported_libs_with_app:
+        return imported_libs_with_app[0], True
+    elif imported_libs:
+        return imported_libs[0], False
+    else:
+        return None, False
+
+
+def asyncio_is_running():
+    """Get whether there is currently a running asyncio loop."""
+    asyncio = sys.modules.get("asyncio", None)
+    if asyncio is None:
+        return False
+    try:
+        loop = asyncio.get_running_loop()
+    except Exception:
+        loop = None
+    return loop is not None
+
+
+# %% Linux window managers
+
+
+SYSTEM_IS_WAYLAND = "wayland" in os.getenv("XDG_SESSION_TYPE", "").lower()
+
+if sys.platform.startswith("linux") and SYSTEM_IS_WAYLAND:
+    # Force glfw to use X11. Note that this does not work if glfw is already imported.
+    if "glfw" not in sys.modules:
+        os.environ["PYGLFW_LIBRARY_VARIANT"] = "x11"
+    # Force Qt to use X11. Qt is more flexible - it ok if e.g. PySide6 is already imported.
+    os.environ["QT_QPA_PLATFORM"] = "xcb"
+    # Force wx to use X11, probably.
+    os.environ["GDK_BACKEND"] = "x11"
+
+
+_x11_display = None
+
+
+def get_alt_x11_display():
+    """Get (the pointer to) a process-global x11 display instance."""
+    # Ideally we'd get the real display object used by the backend.
+    # But this is not always possible. In that case, using an alt display
+    # object can be used.
+    global _x11_display
+    assert sys.platform.startswith("linux")
+    if _x11_display is None:
+        x11 = ctypes.CDLL(ctypes.util.find_library("X11"))
+        x11.XOpenDisplay.restype = ctypes.c_void_p
+        _x11_display = x11.XOpenDisplay(None)
+    return _x11_display
+
+
+_wayland_display = None
+
+
+def get_alt_wayland_display():
+    """Get (the pointer to) a process-global Wayland display instance."""
+    # Ideally we'd get the real display object used by the backend.
+    # This creates a global object, similar to what we do for X11.
+    # Unfortunately, this segfaults, so it looks like the real display object
+    # is needed? Leaving this here for reference.
+    global _wayland_display
+    assert sys.platform.startswith("linux")
+    if _wayland_display is None:
+        wl = ctypes.CDLL(ctypes.util.find_library("wayland-client"))
+        wl.wl_display_connect.restype = ctypes.c_void_p
+        _wayland_display = wl.wl_display_connect(None)
+    return _wayland_display

rendercanvas/_events.py

@@ -1,8 +1,7 @@
 import time
 from collections import defaultdict, deque
 
-from ._gui_utils import log_exception
-from ._coreutils import BaseEnum
+from ._coreutils import log_exception, BaseEnum
 
 
 class EventType(BaseEnum):