diff --git a/pyvda/__init__.py b/pyvda/__init__.py index 9ca598d..65e4a13 100644 --- a/pyvda/__init__.py +++ b/pyvda/__init__.py @@ -62,3 +62,4 @@ def _check_version(): get_virtual_desktops, set_wallpaper_for_all_desktops, ) +from .notification import VirtualDesktopNotificationService diff --git a/pyvda/com_defns.py b/pyvda/com_defns.py index 56121e2..268f6bb 100644 --- a/pyvda/com_defns.py +++ b/pyvda/com_defns.py @@ -9,8 +9,6 @@ * http://grabacr.net/archives/5601 * https://www.cyberforum.ru/blogs/105416/blog3671.html """ -import os -import sys from ctypes import HRESULT, POINTER, c_ulonglong from ctypes.wintypes import ( BOOL, @@ -437,3 +435,104 @@ class IApplicationViewCollection(IUnknown): # STDMETHOD(HRESULT, "RegisterForApplicationViewPositionChanges", (POINTER(IApplicationViewChangeListener), POINTER(DWORD),)), STDMETHOD(HRESULT, "UnregisterForApplicationViewChanges", (DWORD,)), ] + + +# IVirtualDesktopNotificationService - stable across all builds +class IVirtualDesktopNotificationService(IUnknown): + _iid_ = const.GUID_IVirtualDesktopNotificationService + _methods_ = [ + COMMETHOD( + [], + HRESULT, + "Register", + (["in"], LPVOID, "pNotification"), + (["out"], POINTER(DWORD), "pdwCookie"), + ), + COMMETHOD([], HRESULT, "Unregister", (["in"], DWORD, "dwCookie")), + ] + + + +# IVirtualDesktopNotification - build-dependent GUID and vtable +# All parameters use LPVOID because callers should use pyvda's public API +# (VirtualDesktop, AppView) to query state, not unwrap raw COM pointers. + +def _make_notification_iface(guid, methods_spec): + """Create an IVirtualDesktopNotification comtypes interface class. + + Parameters + ---------- + guid : GUID + The COM interface GUID for the current Windows build. + methods_spec : list[tuple[str, int]] + Each entry is (method_name, param_count). Parameters are declared + as ``c_void_p`` since we don't inspect them. + """ + from ctypes import c_void_p + methods = [] + for name, params in methods_spec: + param_defs = [(["in"], c_void_p, f"p{i}") for i in range(params)] + methods.append(COMMETHOD([], HRESULT, name, *param_defs)) + + ns = { + "_iid_": guid, + "_methods_": methods, + } + return type("IVirtualDesktopNotification", (IUnknown,), ns) + + +_NOTIFICATION_METHODS_WIN10 = [ + ("VirtualDesktopCreated", 1), + ("VirtualDesktopDestroyBegin", 2), + ("VirtualDesktopDestroyFailed", 2), + ("VirtualDesktopDestroyed", 2), + ("ViewVirtualDesktopChanged", 1), + ("CurrentVirtualDesktopChanged", 2), +] + +_NOTIFICATION_METHODS_21H2 = [ + ("VirtualDesktopCreated", 2), + ("VirtualDesktopDestroyBegin", 3), + ("VirtualDesktopDestroyFailed", 3), + ("VirtualDesktopDestroyed", 3), + ("Proc7", 1), + ("VirtualDesktopMoved", 4), + ("VirtualDesktopRenamed", 2), + ("ViewVirtualDesktopChanged", 1), + ("CurrentVirtualDesktopChanged", 3), + ("VirtualDesktopWallpaperChanged", 2), +] + +_NOTIFICATION_METHODS_23H2 = [ + ("VirtualDesktopCreated", 1), + ("VirtualDesktopDestroyBegin", 2), + ("VirtualDesktopDestroyFailed", 2), + ("VirtualDesktopDestroyed", 2), + ("VirtualDesktopMoved", 3), + ("VirtualDesktopRenamed", 2), + ("ViewVirtualDesktopChanged", 1), + ("CurrentVirtualDesktopChanged", 2), + ("VirtualDesktopWallpaperChanged", 2), + ("VirtualDesktopSwitched", 1), + ("RemoteVirtualDesktopConnected", 1), +] + +_NOTIFICATION_METHODS_24H2 = _NOTIFICATION_METHODS_23H2 + + +if build.OVER_22631: + GUID_IVirtualDesktopNotification = const.GUID_IVirtualDesktopNotification_22631 + _NOTIFICATION_METHODS = _NOTIFICATION_METHODS_24H2 +elif build.OVER_22621: + GUID_IVirtualDesktopNotification = const.GUID_IVirtualDesktopNotification_22621 + _NOTIFICATION_METHODS = _NOTIFICATION_METHODS_23H2 +elif build.OVER_20231: + GUID_IVirtualDesktopNotification = const.GUID_IVirtualDesktopNotification_20231 + _NOTIFICATION_METHODS = _NOTIFICATION_METHODS_21H2 +else: + GUID_IVirtualDesktopNotification = const.GUID_IVirtualDesktopNotification_9000 + _NOTIFICATION_METHODS = _NOTIFICATION_METHODS_WIN10 + +IVirtualDesktopNotification = _make_notification_iface( + GUID_IVirtualDesktopNotification, _NOTIFICATION_METHODS +) diff --git a/pyvda/const.py b/pyvda/const.py index 31b24f6..bc93478 100644 --- a/pyvda/const.py +++ b/pyvda/const.py @@ -4,6 +4,7 @@ CLSID_VirtualDesktopManagerInternal = GUID("{C5E0CDCA-7B6E-41B2-9FC4-D93975CC467B}") CLSID_IVirtualDesktopManager = GUID("{AA509086-5CA9-4C25-8F95-589D3C07B48A}") CLSID_VirtualDesktopPinnedApps = GUID("{B5A399E7-1C87-46B8-88E9-FC5747B171BD}") +CLSID_IVirtualNotificationService = GUID("{A501FDEC-4A09-464C-AE4E-1B9C21B84918}") GUID_IVirtualDesktop_26100 = GUID("{3F07F4BE-B107-441A-AF0F-39D82529072C}") GUID_IVirtualDesktop_22631 = GUID("{3F07F4BE-B107-441A-AF0F-39D82529072C}") @@ -19,3 +20,12 @@ GUID_IVirtualDesktopManagerInternal_21313 = GUID("{B2F925B9-5A0F-4D2E-9F4D-2B1507593C10}") GUID_IVirtualDesktopManagerInternal_20231 = GUID("{094AFE11-44F2-4BA0-976F-29A97E263EE0}") GUID_IVirtualDesktopManagerInternal_9000 = GUID("{F31574D6-B682-4CDC-BD56-1827860ABEC6}") + +# IVirtualDesktopNotification +GUID_IVirtualDesktopNotification_22631 = GUID("{B9E5E94D-233E-49AB-AF5C-2B4541C3AADE}") +GUID_IVirtualDesktopNotification_22621 = GUID("{B287FA1C-7771-471A-A2DF-9B6B21F0D675}") +GUID_IVirtualDesktopNotification_20231 = GUID("{CD403E52-DEED-4C13-B437-B98380F2B1E8}") +GUID_IVirtualDesktopNotification_9000 = GUID("{C179334C-4295-40D3-BEA1-C654D965605A}") + +# IVirtualDesktopNotificationService stable across all builds +GUID_IVirtualDesktopNotificationService = GUID("{0CD45E71-D927-4F15-8B0A-8FEF525337BF}") diff --git a/pyvda/notification.py b/pyvda/notification.py new file mode 100644 index 0000000..d2b68b5 --- /dev/null +++ b/pyvda/notification.py @@ -0,0 +1,168 @@ +""" +Virtual Desktop Notification Support +===================================== + +Provides COM-based event notifications for virtual desktop changes. + +Usage +----- +.. code:: python + + from pyvda.notification import VirtualDesktopNotificationService + + service = VirtualDesktopNotificationService() + + class MyHandler: + def desktop_changed(self, *args): + print("Desktop changed!") + def desktop_created(self, *args): + print("Desktop created!") + def desktop_destroyed(self, *args): + print("Desktop destroyed!") + + cookie = service.register(MyHandler()) + # ... later ... + service.unregister(cookie) +""" + +import ctypes +import logging +from ctypes import POINTER, c_void_p + +import comtypes + +from pyvda.com_base import IServiceProvider +from pyvda.com_defns import ( + CLSID_ImmersiveShell, + IVirtualDesktopNotification, + IVirtualDesktopNotificationService, +) +from pyvda.const import CLSID_IVirtualNotificationService + +logger = logging.getLogger(__name__) + + +class DesktopNotificationSink(comtypes.COMObject): + """comtypes COMObject implementing IVirtualDesktopNotification. + + Delegates each callback to a user-supplied handler object. + Only methods that exist on the handler are called — missing ones + are silently ignored (return S_OK). + """ + + _com_interfaces_ = [IVirtualDesktopNotification] + + def __init__(self, handler=None): + super().__init__() + self._handler = handler + + def _dispatch(self, method_name, *args): + if self._handler is not None: + fn = getattr(self._handler, method_name, None) + if fn is not None: + try: + fn(*args) + except Exception: + logger.exception("Error in notification handler %s", method_name) + return 0 + + def VirtualDesktopCreated(self, *args): + return self._dispatch("desktop_created", *args) + + def VirtualDesktopDestroyBegin(self, *args): + return self._dispatch("desktop_destroy_begin", *args) + + def VirtualDesktopDestroyFailed(self, *args): + return self._dispatch("desktop_destroy_failed", *args) + + def VirtualDesktopDestroyed(self, *args): + return self._dispatch("desktop_destroyed", *args) + + def VirtualDesktopMoved(self, *args): + return self._dispatch("desktop_moved", *args) + + def VirtualDesktopRenamed(self, *args): + return self._dispatch("desktop_renamed", *args) + + def ViewVirtualDesktopChanged(self, *args): + return self._dispatch("view_changed", *args) + + def CurrentVirtualDesktopChanged(self, *args): + return self._dispatch("desktop_changed", *args) + + def VirtualDesktopWallpaperChanged(self, *args): + return self._dispatch("desktop_wallpaper_changed", *args) + + def VirtualDesktopSwitched(self, *args): + return self._dispatch("desktop_switched", *args) + + def RemoteVirtualDesktopConnected(self, *args): + return self._dispatch("remote_desktop_connected", *args) + + def Proc7(self, *args): + return 0 + + +class VirtualDesktopNotificationService: + """High-level wrapper for IVirtualDesktopNotificationService. + + Handles COM service acquisition, sink creation, and registration. + """ + + def __init__(self): + self._service = None + self._sinks: dict[int, DesktopNotificationSink] = {} + self._acquire_service() + + @staticmethod + def _try_init_com(): + try: + comtypes.CoInitializeEx() + except OSError: + pass + + def _acquire_service(self): + self._try_init_com() + pServiceProvider = comtypes.CoCreateInstance( + CLSID_ImmersiveShell, + IServiceProvider, + comtypes.CLSCTX_LOCAL_SERVER, + ) + pNotifService = POINTER(IVirtualDesktopNotificationService)() + pServiceProvider.QueryService( + CLSID_IVirtualNotificationService, + IVirtualDesktopNotificationService._iid_, + pNotifService, + ) + self._service = pNotifService + + def register(self, handler=None) -> int: + """Register a notification handler and return an integer cookie. + + Parameters + ---------- + handler : object, optional + An object with any of these optional methods: + ``desktop_changed``, ``desktop_created``, ``desktop_destroyed``, + ``desktop_destroy_begin``, ``desktop_destroy_failed``, + ``desktop_moved``, ``desktop_renamed``, ``view_changed``, + ``desktop_wallpaper_changed``, ``desktop_switched``, + ``remote_desktop_connected``. + + If None, a no-op sink is registered (useful for keeping COM alive). + """ + sink = DesktopNotificationSink(handler) + raw_ptr = ctypes.cast( + sink._com_pointers_[IVirtualDesktopNotification._iid_], + c_void_p, + ) + cookie = self._service.Register(raw_ptr) + self._sinks[cookie] = sink # prevent GC + logger.info("Registered desktop notification (cookie=%d)", cookie) + return cookie + + def unregister(self, cookie: int): + """Unregister a previously registered notification handler.""" + self._service.Unregister(cookie) + self._sinks.pop(cookie, None) + logger.info("Unregistered desktop notification (cookie=%d)", cookie) diff --git a/pyvda/pyvda.py b/pyvda/pyvda.py index 3ce924d..3c92c06 100644 --- a/pyvda/pyvda.py +++ b/pyvda/pyvda.py @@ -379,10 +379,10 @@ def remove(self, fallback: Optional[VirtualDesktop] = None): Args: fallback (VirtualDesktop, optional): If you are currently on the desktop you pass to this method, focus will be shifted to the desktop passed here. - If no desktop is passed, it will default to the first. + If no desktop is passed, it will default to a different desktop. """ if fallback is None: - fallback = VirtualDesktop(1) + fallback = VirtualDesktop(2 if self.number == 1 else 1) managers.manager_internal.RemoveDesktop(self._virtual_desktop, fallback._virtual_desktop) # type: ignore def go(self, allow_set_foreground: bool = True): diff --git a/tests/test_notifications.py b/tests/test_notifications.py new file mode 100644 index 0000000..5db520f --- /dev/null +++ b/tests/test_notifications.py @@ -0,0 +1,125 @@ +import ctypes +import ctypes.wintypes +import threading +import time + +import pytest +from comtypes import COINIT_MULTITHREADED, CoInitializeEx + +from pyvda import VirtualDesktop, VirtualDesktopNotificationService, get_virtual_desktops + + +def _pump_messages(event, timeout=5): + """Pump Windows messages until *event* is set or *timeout* seconds elapse. + + COM STA callbacks are dispatched via the Windows message queue, so a + message pump is required in non-GUI processes like pytest.""" + user32 = ctypes.windll.user32 + msg = ctypes.wintypes.MSG() + deadline = time.monotonic() + timeout + while not event.is_set() and time.monotonic() < deadline: + while user32.PeekMessageW(ctypes.byref(msg), None, 0, 0, 1): + user32.TranslateMessage(ctypes.byref(msg)) + user32.DispatchMessageW(ctypes.byref(msg)) + time.sleep(0.01) + + +def test_register_unregister(): + """Basic registration and unregistration should succeed.""" + svc = VirtualDesktopNotificationService() + cookie = svc.register() + assert isinstance(cookie, int) + svc.unregister(cookie) + + +def test_register_with_handler(): + """Registration with a handler object should succeed.""" + class Handler: + pass + + svc = VirtualDesktopNotificationService() + cookie = svc.register(Handler()) + svc.unregister(cookie) + + +def test_multiple_registrations(): + """Multiple handlers can be registered and unregistered independently.""" + svc = VirtualDesktopNotificationService() + cookie1 = svc.register() + cookie2 = svc.register() + assert cookie1 != cookie2 + svc.unregister(cookie1) + svc.unregister(cookie2) + + +def test_desktop_changed_callback(): + """Switching desktops should trigger the desktop_changed callback.""" + if len(get_virtual_desktops()) < 2: + pytest.skip("Need at least 2 desktops to test switching") + + event = threading.Event() + original = VirtualDesktop.current() + + class Handler: + def desktop_changed(self, *args): + event.set() + + svc = VirtualDesktopNotificationService() + cookie = svc.register(Handler()) + try: + target = VirtualDesktop(2) if original.number == 1 else VirtualDesktop(1) + target.go() + _pump_messages(event, timeout=5) + assert event.is_set(), "desktop_changed callback was not fired" + finally: + original.go() + time.sleep(0.5) + svc.unregister(cookie) + + +def test_desktop_created_destroyed_callback(): + """Creating and destroying a desktop should trigger callbacks.""" + created_event = threading.Event() + destroyed_event = threading.Event() + + class Handler: + def desktop_created(self, *args): + created_event.set() + + def desktop_destroyed(self, *args): + destroyed_event.set() + + svc = VirtualDesktopNotificationService() + cookie = svc.register(Handler()) + try: + new_desktop = VirtualDesktop.create() + _pump_messages(created_event, timeout=5) + assert created_event.is_set(), "desktop_created callback was not fired" + + new_desktop.remove(fallback=VirtualDesktop(1)) + _pump_messages(destroyed_event, timeout=5) + assert destroyed_event.is_set(), "desktop_destroyed callback was not fired" + time.sleep(1) # Wait for animation + finally: + svc.unregister(cookie) + + +def test_register_unregister_from_thread(): + """Registration should work from a non-main thread.""" + error = None + + def f(): + nonlocal error + try: + CoInitializeEx(COINIT_MULTITHREADED) + svc = VirtualDesktopNotificationService() + cookie = svc.register() + svc.unregister(cookie) + except Exception as e: + error = e + + t = threading.Thread(target=f) + t.start() + t.join() + if error is not None: + raise error