Auto-recover DXGI capture after duplication interface loss

BetterCam/DXcam engines now detect when the DXGI Desktop Duplication
interface is lost (display mode change, sleep/wake, UAC prompt, etc.)
and automatically reinitialize the camera with a 3-second cooldown
between attempts, instead of error-looping indefinitely.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-20 22:47:54 +03:00
parent 0a000cc44c
commit 5004992f26
2 changed files with 54 additions and 0 deletions

View File

@@ -1,6 +1,7 @@
"""BetterCam-based screen capture engine (Windows only, DXGI Desktop Duplication).""" """BetterCam-based screen capture engine (Windows only, DXGI Desktop Duplication)."""
import sys import sys
import time
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
import numpy as np import numpy as np
@@ -15,6 +16,8 @@ from wled_controller.utils import get_logger
logger = get_logger(__name__) logger = get_logger(__name__)
_REINIT_COOLDOWN = 3.0 # seconds between re-init attempts
class BetterCamCaptureStream(CaptureStream): class BetterCamCaptureStream(CaptureStream):
"""BetterCam capture stream for a specific display.""" """BetterCam capture stream for a specific display."""
@@ -23,6 +26,7 @@ class BetterCamCaptureStream(CaptureStream):
super().__init__(display_index, config) super().__init__(display_index, config)
self._camera = None self._camera = None
self._bettercam = None self._bettercam = None
self._last_reinit: float = 0
def initialize(self) -> None: def initialize(self) -> None:
try: try:
@@ -48,6 +52,7 @@ class BetterCamCaptureStream(CaptureStream):
raise RuntimeError(f"Failed to create BetterCam camera for display {self.display_index}") raise RuntimeError(f"Failed to create BetterCam camera for display {self.display_index}")
self._initialized = True self._initialized = True
self._last_reinit = time.monotonic()
logger.info(f"BetterCam capture stream initialized (display={self.display_index})") logger.info(f"BetterCam capture stream initialized (display={self.display_index})")
def cleanup(self) -> None: def cleanup(self) -> None:
@@ -72,6 +77,24 @@ class BetterCamCaptureStream(CaptureStream):
self._initialized = False self._initialized = False
logger.info(f"BetterCam capture stream cleaned up (display={self.display_index})") logger.info(f"BetterCam capture stream cleaned up (display={self.display_index})")
def _try_reinit(self) -> bool:
"""Attempt to reinitialize the DXGI capture after a failure.
Returns True if reinit succeeded, False if on cooldown or failed.
"""
now = time.monotonic()
if now - self._last_reinit < _REINIT_COOLDOWN:
return False
self._last_reinit = now
logger.warning(f"BetterCam: reinitializing camera for display {self.display_index}")
try:
self.cleanup()
self.initialize()
return True
except Exception as reinit_err:
logger.error(f"BetterCam reinit failed (display={self.display_index}): {reinit_err}")
return False
def capture_frame(self) -> Optional[ScreenCapture]: def capture_frame(self) -> Optional[ScreenCapture]:
if not self._initialized: if not self._initialized:
self.initialize() self.initialize()
@@ -98,6 +121,10 @@ class BetterCamCaptureStream(CaptureStream):
raise raise
except Exception as e: except Exception as e:
logger.error(f"Failed to capture display {self.display_index} with BetterCam: {e}") logger.error(f"Failed to capture display {self.display_index} with BetterCam: {e}")
# DXGI duplication can be lost after display changes / sleep / UAC.
# Attempt to recreate the camera so capture can recover.
if self._try_reinit():
return None # let the caller retry on next tick
raise RuntimeError(f"Screen capture failed: {e}") raise RuntimeError(f"Screen capture failed: {e}")

View File

@@ -1,6 +1,7 @@
"""DXcam-based screen capture engine (Windows only, DXGI Desktop Duplication).""" """DXcam-based screen capture engine (Windows only, DXGI Desktop Duplication)."""
import sys import sys
import time
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
import numpy as np import numpy as np
@@ -15,6 +16,8 @@ from wled_controller.utils import get_logger
logger = get_logger(__name__) logger = get_logger(__name__)
_REINIT_COOLDOWN = 3.0 # seconds between re-init attempts
class DXcamCaptureStream(CaptureStream): class DXcamCaptureStream(CaptureStream):
"""DXcam capture stream for a specific display.""" """DXcam capture stream for a specific display."""
@@ -23,6 +26,7 @@ class DXcamCaptureStream(CaptureStream):
super().__init__(display_index, config) super().__init__(display_index, config)
self._camera = None self._camera = None
self._dxcam = None self._dxcam = None
self._last_reinit: float = 0
def initialize(self) -> None: def initialize(self) -> None:
try: try:
@@ -48,6 +52,7 @@ class DXcamCaptureStream(CaptureStream):
raise RuntimeError(f"Failed to create DXcam camera for display {self.display_index}") raise RuntimeError(f"Failed to create DXcam camera for display {self.display_index}")
self._initialized = True self._initialized = True
self._last_reinit = time.monotonic()
logger.info(f"DXcam capture stream initialized (display={self.display_index})") logger.info(f"DXcam capture stream initialized (display={self.display_index})")
def cleanup(self) -> None: def cleanup(self) -> None:
@@ -72,6 +77,24 @@ class DXcamCaptureStream(CaptureStream):
self._initialized = False self._initialized = False
logger.info(f"DXcam capture stream cleaned up (display={self.display_index})") logger.info(f"DXcam capture stream cleaned up (display={self.display_index})")
def _try_reinit(self) -> bool:
"""Attempt to reinitialize the DXGI capture after a failure.
Returns True if reinit succeeded, False if on cooldown or failed.
"""
now = time.monotonic()
if now - self._last_reinit < _REINIT_COOLDOWN:
return False
self._last_reinit = now
logger.warning(f"DXcam: reinitializing camera for display {self.display_index}")
try:
self.cleanup()
self.initialize()
return True
except Exception as reinit_err:
logger.error(f"DXcam reinit failed (display={self.display_index}): {reinit_err}")
return False
def capture_frame(self) -> Optional[ScreenCapture]: def capture_frame(self) -> Optional[ScreenCapture]:
if not self._initialized: if not self._initialized:
self.initialize() self.initialize()
@@ -98,6 +121,10 @@ class DXcamCaptureStream(CaptureStream):
raise raise
except Exception as e: except Exception as e:
logger.error(f"Failed to capture display {self.display_index} with DXcam: {e}") logger.error(f"Failed to capture display {self.display_index} with DXcam: {e}")
# DXGI duplication can be lost after display changes / sleep / UAC.
# Attempt to recreate the camera so capture can recover.
if self._try_reinit():
return None # let the caller retry on next tick
raise RuntimeError(f"Screen capture failed: {e}") raise RuntimeError(f"Screen capture failed: {e}")