Create WindowsVCP for each physical monitor instead of logical

monitorcontrol/vcp/vcp_windows.py

@@ -1,6 +1,6 @@
 from .vcp_abc import VCP, VCPError
 from types import TracebackType
-from typing import List, Optional, Tuple, Type
+from typing import Iterator, List, Optional, Tuple, Type
 import ctypes
 import logging
 import sys
@@ -31,68 +31,28 @@ class WindowsVCP(VCP):
             https://stackoverflow.com/questions/16588133/
         """
 
-        def __init__(self, hmonitor: HMONITOR):
+        def __init__(self, handle: HANDLE, description: str):
             """
             Args:
-                hmonitor: logical monitor handle
+                handle: Handle to the physical monitor.
+                description: Text description of the physical monitor.
             """
             self.logger = logging.getLogger(__name__)
-            self.hmonitor = hmonitor
+            self.handle = handle
+            self.description = description
 
-        def __enter__(self):
-            num_physical = DWORD()
-            self.logger.debug("GetNumberOfPhysicalMonitorsFromHMONITOR")
-            try:
-                if not ctypes.windll.dxva2.GetNumberOfPhysicalMonitorsFromHMONITOR(
-                    self.hmonitor, ctypes.byref(num_physical)
-                ):
-                    raise VCPError(
-                        "Call to GetNumberOfPhysicalMonitorsFromHMONITOR failed: "
-                        + ctypes.FormatError()
-                    )
-            except OSError as e:
-                raise VCPError(
-                    "Call to GetNumberOfPhysicalMonitorsFromHMONITOR failed"
-                ) from e
-
-            if num_physical.value == 0:
-                raise VCPError("no physical monitor found")
-            elif num_physical.value > 1:
-                # TODO: Figure out a clever way around the Windows API since
-                # it does not allow opening and closing of individual physical
-                # monitors without their hmonitors.
-                raise VCPError("more than one physical monitor per hmonitor")
+        def __del__(self):
+            WindowsVCP._destroy_physical_monitor(self.handle)
 
-            physical_monitors = (PhysicalMonitor * num_physical.value)()
-            self.logger.debug("GetPhysicalMonitorsFromHMONITOR")
-            try:
-                if not ctypes.windll.dxva2.GetPhysicalMonitorsFromHMONITOR(
-                    self.hmonitor, num_physical.value, physical_monitors
-                ):
-                    raise VCPError(
-                        "Call to GetPhysicalMonitorsFromHMONITOR failed: "
-                        + ctypes.FormatError()
-                    )
-            except OSError as e:
-                raise VCPError("failed to open physical monitor handle") from e
-            self.handle = physical_monitors[0].handle
-            self.description = physical_monitors[0].description
-            return self
+        def __enter__(self):
+            pass
 
         def __exit__(
             self,
             exception_type: Optional[Type[BaseException]],
             exception_value: Optional[BaseException],
             exception_traceback: Optional[TracebackType],
         ) -> Optional[bool]:
-            self.logger.debug("DestroyPhysicalMonitor")
-            try:
-                if not ctypes.windll.dxva2.DestroyPhysicalMonitor(self.handle):
-                    raise VCPError(
-                        "Call to DestroyPhysicalMonitor failed: " + ctypes.FormatError()
-                    )
-            except OSError as e:
-                raise VCPError("failed to close handle") from e
             return False
 
         def set_vcp_feature(self, code: int, value: int):
@@ -190,6 +150,92 @@ def get_vcp_capabilities(self):
                 raise VCPError("failed to get VCP capabilities") from e
             return cap_string.value.decode("ascii")
 
+        @staticmethod
+        def _get_physical_monitors() -> Iterator[Tuple[HANDLE, str]]:
+            """
+            Returns a list of physical monitors.
+            """
+            return (
+                physical_monitor
+                for hmonitor in WindowsVCP._get_hmonitors()
+                for physical_monitor in WindowsVCP._physical_monitors_from_hmonitor(
+                    hmonitor
+                )
+            )
+
+        @staticmethod
+        def _get_hmonitors() -> List[HMONITOR]:
+            """
+            Calls the Windows `EnumDisplayMonitors` API in Python-friendly form.
+            """
+            hmonitors = []  # type: List[HMONITOR]
+            try:
+
+                def _callback(hmonitor, hdc, lprect, lparam):
+                    hmonitors.append(HMONITOR(hmonitor))
+                    del hmonitor, hdc, lprect, lparam
+                    return True  # continue enumeration
+
+                MONITORENUMPROC = ctypes.WINFUNCTYPE(  # noqa: N806
+                    BOOL, HMONITOR, HDC, ctypes.POINTER(RECT), LPARAM
+                )
+                callback = MONITORENUMPROC(_callback)
+                if not ctypes.windll.user32.EnumDisplayMonitors(0, 0, callback, 0):
+                    raise VCPError("Call to EnumDisplayMonitors failed")
+            except OSError as e:
+                raise VCPError("failed to enumerate VCPs") from e
+            return hmonitors
+
+        @staticmethod
+        def _physical_monitors_from_hmonitor(
+            hmonitor: HMONITOR,
+        ) -> Iterator[Tuple[HANDLE, str]]:
+            """
+            Calls the Windows `GetPhysicalMonitorsFromHMONITOR` API in Python-friendly form.
+            """
+            num_physical = DWORD()
+            try:
+                if not ctypes.windll.dxva2.GetNumberOfPhysicalMonitorsFromHMONITOR(
+                    hmonitor, ctypes.byref(num_physical)
+                ):
+                    raise VCPError(
+                        "Call to GetNumberOfPhysicalMonitorsFromHMONITOR failed: "
+                        + ctypes.FormatError()
+                    )
+            except OSError as e:
+                raise VCPError(
+                    "Call to GetNumberOfPhysicalMonitorsFromHMONITOR failed"
+                ) from e
+
+            physical_monitors = (PhysicalMonitor * num_physical.value)()
+            try:
+                if not ctypes.windll.dxva2.GetPhysicalMonitorsFromHMONITOR(
+                    hmonitor, num_physical.value, physical_monitors
+                ):
+                    raise VCPError(
+                        "Call to GetPhysicalMonitorsFromHMONITOR failed: "
+                        + ctypes.FormatError()
+                    )
+            except OSError as e:
+                raise VCPError("failed to open physical monitor handle") from e
+            return (
+                [physical_monitor.handle, physical_monitor.description]
+                for physical_monitor in physical_monitors
+            )
+
+        @staticmethod
+        def _destroy_physical_monitor(handle: HANDLE) -> None:
+            """
+            Calls the Windows `DestroyPhysicalMonitor` API in Python-friendly form.
+            """
+            try:
+                if not ctypes.windll.dxva2.DestroyPhysicalMonitor(handle):
+                    raise VCPError(
+                        "Call to DestroyPhysicalMonitor failed: " + ctypes.FormatError()
+                    )
+            except OSError as e:
+                raise VCPError("failed to close handle") from e
+
     def get_vcps() -> List[WindowsVCP]:
         """
         Opens handles to all physical VCPs.
@@ -200,26 +246,8 @@ def get_vcps() -> List[WindowsVCP]:
         Raises:
             VCPError: Failed to enumerate VCPs.
         """
-        vcps = []
-        hmonitors = []
-
-        try:
-
-            def _callback(hmonitor, hdc, lprect, lparam):
-                hmonitors.append(HMONITOR(hmonitor))
-                del hmonitor, hdc, lprect, lparam
-                return True  # continue enumeration
-
-            MONITORENUMPROC = ctypes.WINFUNCTYPE(  # noqa: N806
-                BOOL, HMONITOR, HDC, ctypes.POINTER(RECT), LPARAM
-            )
-            callback = MONITORENUMPROC(_callback)
-            if not ctypes.windll.user32.EnumDisplayMonitors(0, 0, callback, 0):
-                raise VCPError("Call to EnumDisplayMonitors failed")
-        except OSError as e:
-            raise VCPError("failed to enumerate VCPs") from e
-
-        for logical in hmonitors:
-            vcps.append(WindowsVCP(logical))
-
-        return vcps
+        physical_monitors = WindowsVCP._get_physical_monitors()
+        return list(
+            WindowsVCP(handle, description)
+            for (handle, description) in physical_monitors
+        )

tests/test_windows_vcp.py

@@ -0,0 +1,33 @@
+import pytest
+import sys
+from unittest.mock import patch
+
+
+if sys.platform == "win32":
+    from monitorcontrol.vcp.vcp_windows import WindowsVCP
+
+    @pytest.mark.parametrize(
+        "monitor_input, expected",
+        [
+            [[1], ["1-0"]],
+            [[2], ["2-0", "2-1"]],
+            [[1, 2], ["1-0", "2-0", "2-1"]],
+            [[1, 3], ["1-0", "3-0", "3-1", "3-2"]],
+        ],
+    )
+    @patch("monitorcontrol.vcp.vcp_windows.WindowsVCP._get_hmonitors")
+    @patch("monitorcontrol.vcp.vcp_windows.WindowsVCP._physical_monitors_from_hmonitor")
+    def test_get_physical_monitors(
+        physical_monitors_from_hmonitor, get_hmonitors, monitor_input, expected
+    ):
+        get_hmonitors.return_value = monitor_input
+        physical_monitors = {
+            1: ["1-0"],
+            2: ["2-0", "2-1"],
+            3: ["3-0", "3-1", "3-2"],
+        }
+        physical_monitors_from_hmonitor.side_effect = (
+            lambda hmonitor: physical_monitors.get(hmonitor)
+        )
+        result = list(WindowsVCP._get_physical_monitors())
+        assert result == expected