Robustness: Handle stale COM objects and RPC server unavailability

pyvda/const.py

@@ -19,3 +19,8 @@
 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}")
+
+RPC_S_SERVER_UNAVAILABLE = -2147023174
+RPC_E_DISCONNECTED = -2147417848
+RPC_S_SERVER_UNAVAILABLE_U = 2147944122
+RPC_E_DISCONNECTED_U = 2147549432

pyvda/pyvda.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import functools
+import time
 from ctypes import windll
 from typing import List, Optional
 
@@ -8,6 +10,12 @@
 
 import pyvda.build as build
 from pyvda.com_defns import IApplicationView, IVirtualDesktop, IVirtualDesktop2
+from pyvda.const import (
+    RPC_E_DISCONNECTED,
+    RPC_E_DISCONNECTED_U,
+    RPC_S_SERVER_UNAVAILABLE,
+    RPC_S_SERVER_UNAVAILABLE_U,
+)
 from pyvda.utils import Managers
 from pyvda.winstring import HSTRING
 
@@ -16,6 +24,32 @@
 
 managers = Managers()
 
+def _com_retry(func):
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        last_error = None
+        for i in range(3):
+            try:
+                return func(*args, **kwargs)
+            except _ctypes.COMError as e:
+                if e.args and e.args[0] in (
+                    RPC_S_SERVER_UNAVAILABLE,
+                    RPC_E_DISCONNECTED,
+                    RPC_S_SERVER_UNAVAILABLE_U,
+                    RPC_E_DISCONNECTED_U,
+                ):
+                    last_error = e
+                    time.sleep(0.1 * (i + 1))
+                    if args and hasattr(args[0], '_refresh') and not isinstance(args[0], type) and func.__name__ != '__init__':
+                        args[0]._refresh()
+                    else:
+                        managers.__init__()
+                    continue
+                raise
+        if last_error:
+            raise last_error
+    return wrapper
+
 
 class AppView():
     """
@@ -27,6 +61,7 @@ class AppView():
 
     """
 
+    @_com_retry
     def __init__(self, hwnd: Optional[int] = None, view: Optional['IApplicationView'] = None):
         """One of the following parameters must be provided:
 
@@ -41,17 +76,25 @@ def __init__(self, hwnd: Optional[int] = None, view: Optional['IApplicationView'
             self._view = view
         else:
             raise Exception(f"Must pass 'hwnd' or 'view'")
+
+        self._hwnd_cached = self._view.GetThumbnailWindow()
+
+    def _refresh(self):
+        managers.__init__()
+        self._view = managers.view_collection.GetViewForHwnd(self._hwnd_cached)
 
     def __eq__(self, other):
         return self.hwnd == other.hwnd
 
     @property
+    @_com_retry
     def hwnd(self) -> int:
         """This window's handle.
         """
         return self._view.GetThumbnailWindow() # type: ignore
 
     @property
+    @_com_retry
     def app_id(self) -> Optional[int]:
         """The ID of this window's app. Some specific types of windows do not have an app ID, and will return `None`.
         """
@@ -60,10 +103,18 @@ def app_id(self) -> Optional[int]:
             # This seems to happen for things like window managers which are pinned above the normal windows.
             # Can be reliably reproduced with the 'f.lux' options window.
             return self._view.GetAppUserModelId() # type: ignore
-        except _ctypes.COMError:
+        except _ctypes.COMError as e:
+            if e.args and e.args[0] in (
+                RPC_S_SERVER_UNAVAILABLE,
+                RPC_E_DISCONNECTED,
+                RPC_S_SERVER_UNAVAILABLE_U,
+                RPC_E_DISCONNECTED_U,
+            ):
+                raise
             return None
 
     @classmethod
+    @_com_retry
     def current(cls):
         """
         Returns:
@@ -75,25 +126,30 @@ def current(cls):
     #  ------------------------------------------------
     #  IApplicationView methods
     #  ------------------------------------------------
+    @_com_retry
     def is_shown_in_switchers(self) -> bool:
         """Is the view shown in the alt-tab view?
         """
         return bool(self._view.GetShowInSwitchers()) # type: ignore
 
+    @_com_retry
     def is_visible(self) -> bool:
         """Is the view visible?
         """
         return bool(self._view.GetVisibility()) # type: ignore
 
+    @_com_retry
     def get_activation_timestamp(self) -> int:
         """Get the last activation timestamp for this window.
         """
         return self._view.GetLastActivationTimestamp() # type: ignore
 
+    @_com_retry
     def set_focus(self):
         """Focus the window"""
         return self._view.SetFocus() # type: ignore
 
+    @_com_retry
     def switch_to(self):
         """Switch to the window. Behaves slightly differently to set_focus -
         this is what is called when you use the alt-tab menu."""
@@ -103,18 +159,21 @@ def switch_to(self):
     #  ------------------------------------------------
     #  IVirtualDesktopPinnedApps methods
     #  ------------------------------------------------
+    @_com_retry
     def pin(self):
         """
         Pin the window (corresponds to the 'show window on all desktops' toggle).
         """
         managers.pinned_apps.PinView(self._view) # type: ignore
 
+    @_com_retry
     def unpin(self):
         """
         Unpin the window (corresponds to the 'show window on all desktops' toggle).
         """
         managers.pinned_apps.UnpinView(self._view) # type: ignore
 
+    @_com_retry
     def is_pinned(self) -> bool:
         """
         Check if this window is pinned (corresponds to the 'show window on all desktops' toggle).
@@ -124,6 +183,7 @@ def is_pinned(self) -> bool:
         """
         return managers.pinned_apps.IsViewPinned(self._view) # type: ignore
 
+    @_com_retry
     def pin_app(self):
         """
         Pin this window's app (corresponds to the 'show windows from this app on all desktops' toggle).
@@ -135,6 +195,7 @@ def pin_app(self):
             return
         managers.pinned_apps.PinAppID(self.app_id) # type: ignore
 
+    @_com_retry
     def unpin_app(self):
         """
         Unpin this window's app (corresponds to the 'show windows from this app on all desktops' toggle).
@@ -144,6 +205,7 @@ def unpin_app(self):
             return
         managers.pinned_apps.UnpinAppID(self.app_id) # type: ignore
 
+    @_com_retry
     def is_app_pinned(self) -> bool:
         """
         Check if this window's app is pinned (corresponds to the 'show windows from this app on all desktops' toggle).
@@ -160,6 +222,7 @@ def is_app_pinned(self) -> bool:
     #  ------------------------------------------------
     #  IVirtualDesktopManagerInternal methods
     #  ------------------------------------------------
+    @_com_retry
     def move(self, desktop: VirtualDesktop):
         """Move the window to a different virtual desktop.
 
@@ -174,6 +237,7 @@ def move(self, desktop: VirtualDesktop):
         managers.manager_internal.MoveViewToDesktop(self._view, desktop._virtual_desktop)  # type: ignore
 
     @property
+    @_com_retry
     def desktop_id(self) -> GUID:
         """
         Returns:
@@ -182,14 +246,15 @@ def desktop_id(self) -> GUID:
         return self._view.GetVirtualDesktopId() # type: ignore
 
     @property
+    @_com_retry
     def desktop(self) -> VirtualDesktop:
         """
         Returns:
             VirtualDesktop: The virtual desktop which this window is on.
         """
         return VirtualDesktop(desktop_id=self.desktop_id)
 
-
+    @_com_retry
     def is_on_desktop(self, desktop: VirtualDesktop, include_pinned: bool = True) -> bool:
         """Is this window on the passed virtual desktop?
 
@@ -207,13 +272,14 @@ def is_on_desktop(self, desktop: VirtualDesktop, include_pinned: bool = True) ->
         else:
             return self.desktop_id == desktop.id
 
-
+    @_com_retry
     def is_on_current_desktop(self) -> bool:
         """Is this window on the current desktop?
         """
         return self.is_on_desktop(VirtualDesktop.current())
 
 
+@_com_retry
 def get_apps_by_z_order(switcher_windows: bool = True, current_desktop: bool = True) -> List[AppView]:
     """Get a list of AppViews, ordered by their Z position, with
     the foreground window first.
@@ -226,12 +292,12 @@ def get_apps_by_z_order(switcher_windows: bool = True, current_desktop: bool = T
         List[AppView]: AppViews matching the specified criteria.
     """
     views_arr = managers.view_collection.GetViewsByZOrder() # type: ignore
-    all_views = [AppView(view=v) for v in views_arr.iter(IApplicationView)]
+    all_views =[AppView(view=v) for v in views_arr.iter(IApplicationView)]
     if not switcher_windows and not current_desktop:
         # no filters
         return all_views
 
-    result = []
+    result =[]
     vd = VirtualDesktop.current()
     for view in all_views:
         if switcher_windows and not view.is_shown_in_switchers():
@@ -246,6 +312,7 @@ class VirtualDesktop():
     """
     Wrapper around the `IVirtualDesktop` COM object, representing one virtual desktop.
     """
+    @_com_retry
     def __init__(
         self,
         number: Optional[int] = None,
@@ -284,8 +351,15 @@ def __init__(
 
         else:
             raise Exception("Must provide one of 'number', 'desktop_id' or 'desktop'")
+
+        self._id_cached = self._virtual_desktop.GetID()
+
+    def _refresh(self):
+        managers.__init__()
+        self._virtual_desktop = managers.manager_internal.FindDesktop(self._id_cached)
 
     @classmethod
+    @_com_retry
     def current(cls):
         """Convenience method to return a `VirtualDesktop` object for the
         currently active desktop.
@@ -296,6 +370,7 @@ def current(cls):
         return cls(current=True)
 
     @classmethod
+    @_com_retry
     def create(cls):
         """Create a new virtual desktop.
 
@@ -306,6 +381,7 @@ def create(cls):
         return cls(desktop=desktop)
 
     @property
+    @_com_retry
     def id(self) -> GUID:
         """The GUID of this desktop.
 
@@ -315,6 +391,7 @@ def id(self) -> GUID:
         return self._virtual_desktop.GetID() # type: ignore
 
     @property
+    @_com_retry
     def number(self) -> int:
         """The index of this virtual desktop in the task view. Between 1 and
         the total number of desktops active.
@@ -330,6 +407,7 @@ def number(self) -> int:
             raise Exception(f"Desktop with ID {self.id} not found")
 
     @property
+    @_com_retry
     def name(self) -> str:
         """The name of this virtual desktop in the task view.
         Note that the default name is an empty string even though the task view shows
@@ -354,6 +432,7 @@ def name(self) -> str:
         else:
             raise Exception(f"Desktop with ID {self.id} not found")
 
+    @_com_retry
     def rename(self, name: str):
         """Rename this desktop.
 
@@ -373,6 +452,7 @@ def rename(self, name: str):
 
         managers.manager_internal.SetName(self._virtual_desktop, HSTRING(name)) # type: ignore
 
+    @_com_retry
     def remove(self, fallback: Optional[VirtualDesktop] = None):
         """Delete this virtual desktop, falling back to 'fallback'.
 
@@ -385,6 +465,7 @@ def remove(self, fallback: Optional[VirtualDesktop] = None):
             fallback = VirtualDesktop(1)
         managers.manager_internal.RemoveDesktop(self._virtual_desktop, fallback._virtual_desktop) # type: ignore
 
+    @_com_retry
     def go(self, allow_set_foreground: bool = True):
         """Switch to this virtual desktop.
 
@@ -398,6 +479,7 @@ def go(self, allow_set_foreground: bool = True):
             windll.user32.AllowSetForegroundWindow(ASFW_ANY)
         managers.manager_internal.switch_desktop(self._virtual_desktop) # type: ignore
 
+    @_com_retry
     def apps_by_z_order(self, include_pinned: bool = True) -> List[AppView]:
         """Get a list of AppViews, ordered by their Z position, with
         the foreground window first.
@@ -410,13 +492,14 @@ def apps_by_z_order(self, include_pinned: bool = True) -> List[AppView]:
             List[AppView]: AppViews matching the specified criteria.
         """
         views_arr = managers.view_collection.GetViewsByZOrder() # type: ignore
-        all_views = [AppView(view=v) for v in views_arr.iter(IApplicationView)]
-        result = []
+        all_views =[AppView(view=v) for v in views_arr.iter(IApplicationView)]
+        result =[]
         for view in all_views:
             if view.is_shown_in_switchers() and view.is_on_desktop(self, include_pinned):
                 result.append(view)
         return result
 
+    @_com_retry
     def set_wallpaper(self, path: str):
         """Set wallpaper on current virtual desktop to `path`.
 
@@ -429,16 +512,18 @@ def set_wallpaper(self, path: str):
             raise NotImplementedError("set_wallpaper is only available on Windows 11")
 
 
+@_com_retry
 def get_virtual_desktops() -> List[VirtualDesktop]:
     """Return a list of all current virtual desktops, one for each desktop visible in the task view.
 
     Returns:
         List[VirtualDesktop]: Virtual desktops currently active.
     """
     array = managers.manager_internal.get_all_desktops() # type: ignore
-    return [VirtualDesktop(desktop=vd) for vd in array.iter(IVirtualDesktop)]
+    return[VirtualDesktop(desktop=vd) for vd in array.iter(IVirtualDesktop)]
 
 
+@_com_retry
 def set_wallpaper_for_all_desktops(path: str):
     """Set wallpaper on current virtual desktop to `path`.
 
@@ -448,4 +533,4 @@ def set_wallpaper_for_all_desktops(path: str):
     if build.OVER_21313:
         managers.manager_internal.SetWallpaperForAllDesktops(path=HSTRING(path)) # type: ignore
     else:
-        raise NotImplementedError("set_wallpaper_for_all_desktops is only available on Windows 11")
+        raise NotImplementedError("set_wallpaper_for_all_desktops is only available on Windows 11")