Refactor DDM backend to use monitorcontrol for input switching and update error messages for clarity

README.md

@@ -22,7 +22,7 @@ Samsung trigger input codes mapped by port:
 - `DP2` -> `19`
 - `HDMI` -> `17`
 
-- If Samsung trigger input matches this device port code, the app runs `DDM.exe /MNT:AW3423DWF /WriteActiveInput <PORT>`.
+- If Samsung trigger input matches this device port code, the app runs `monitor.set_input_source(<PORT>)` on `AW3423DWF` via `monitorcontrol`.
 - If it does not match, the app waits.
 
 Retry behavior:

app/ddm.py

@@ -1,18 +1,14 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from pathlib import Path
 from threading import Lock
-import shutil
-import subprocess
-import tempfile
-import time
 from typing import Protocol
 
+from monitorcontrol import get_monitors
+
 from app.config import SUPPORTED_TARGET_PORTS
 
 
-DEFAULT_DDM_PATH = Path(r"C:\Program Files\Dell\Dell Display Manager 2.0\DDM.exe")
 ALIENWARE_MODEL_TOKEN = "AW3423DWF"
 
 
@@ -38,20 +34,15 @@ class DDMBackend(Protocol):
 class RealDDMBackend:
     def __init__(
         self,
-        executable_path: Path | None = None,
+        alienware_model_token: str = ALIENWARE_MODEL_TOKEN,
-        max_slots: int = 8,
-        command_timeout_seconds: float = 4.0,
-        log_timeout_seconds: float = 3.0,
     ):
-        self.executable_path = executable_path or self._find_executable()
+        self.alienware_model_token = alienware_model_token
-        self.max_slots = max_slots
-        self.command_timeout_seconds = command_timeout_seconds
-        self.log_timeout_seconds = log_timeout_seconds
         self._cached_slot: int | None = None
         self._lock = Lock()
 
     def is_available(self) -> bool:
-        return self.executable_path is not None and self.executable_path.exists()
+        _, error = self._resolve_alienware_monitor()
+        return error is None
 
     def supports_monitor_targeting(self) -> bool:
         return True
@@ -61,143 +52,72 @@ class RealDDMBackend:
             self._cached_slot = None
 
     def resolve_alienware_slot(self, force: bool = False) -> int | None:
-        if not self.is_available():
-            return None
-
         with self._lock:
             if self._cached_slot is not None and not force:
                 return self._cached_slot
 
-            for slot in range(1, self.max_slots + 1):
+            monitor, error = self._resolve_alienware_monitor()
-                result = self._run_logged_command(f"/{slot}:ReadAssetAttributes")
+            del monitor
-                line = _extract_result_line(result, f"{slot}:ReadAssetAttributes")
+            if error is not None:
-                if not line:
+                self._cached_slot = None
-                    continue
+                return None
-                if "INVALID COMMAND" in line.upper():
+            # Service still expects a slot-like value. With monitorcontrol we
-                    continue
+            # target by model, so we expose a synthetic stable slot.
-                if ALIENWARE_MODEL_TOKEN in line.upper():
+            self._cached_slot = 1
-                    self._cached_slot = slot
+            return self._cached_slot
-                    return slot
-
-            self._cached_slot = None
-            return None
 
     def switch_to_port(self, slot: int, port_name: str) -> DDMCommandResult:
-        if not self.is_available():
+        del slot
-            return DDMCommandResult(False, "DDM.exe was not found.")
-
         port_spec = SUPPORTED_TARGET_PORTS.get(port_name.upper())
         if not port_spec:
             return DDMCommandResult(False, f"Port {port_name} is not supported.")
 
-        ddm_input = str(port_spec["ddm_input"])
+        target_input = _to_monitor_input_value(str(port_spec["ddm_input"]))
-        monitor_selector = f"/MNT:{ALIENWARE_MODEL_TOKEN}"
+        monitor, error = self._resolve_alienware_monitor()
-
+        if error is not None:
-        targeted_output = self._run_logged_command(
+            return DDMCommandResult(False, error)
-            monitor_selector,
+        assert monitor is not None
-            "/WriteActiveInput",
-            ddm_input,
-        )
-        targeted_line = _extract_result_line_fragment(targeted_output, "WRITEACTIVEINPUT")
-        if not _output_has_failure(targeted_line, targeted_output):
-            return DDMCommandResult(
-                True,
-                f"Alienware {ALIENWARE_MODEL_TOKEN} was instructed to switch to {ddm_input}.",
-                targeted_output,
-            )
-
-        if slot > 0:
-            legacy_output = self._run_logged_command(f"/{slot}:WriteActiveInput", ddm_input)
-            legacy_line = _extract_result_line(legacy_output, f"{slot}:WriteActiveInput")
-            if not _output_has_failure(legacy_line, legacy_output):
-                return DDMCommandResult(
-                    True,
-                    f"Alienware DDM slot {slot} was instructed to switch to {ddm_input}.",
-                    legacy_output,
-                )
-            merged_output = "\n".join(
-                block
-                for block in [
-                    "[targeted]",
-                    targeted_output.strip(),
-                    "[legacy]",
-                    legacy_output.strip(),
-                ]
-                if block
-            )
-            return DDMCommandResult(False, "DDM rejected the WriteActiveInput command.", merged_output)
-
-        return DDMCommandResult(False, "DDM rejected the WriteActiveInput command.", targeted_output)
-
-    def _run_logged_command(self, *command_args: str) -> str:
-        assert self.executable_path is not None
-
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as handle:
-            log_path = Path(handle.name)
 
         try:
-            subprocess.run(
+            with monitor:
-                [str(self.executable_path), "/Log", str(log_path), *command_args],
+                monitor.set_input_source(target_input)
-                check=False,
+        except Exception as exc:
-                capture_output=True,
+            return DDMCommandResult(
-                text=True,
+                False,
-                timeout=self.command_timeout_seconds,
+                f"monitorcontrol failed to switch Alienware input to {target_input}: {exc}",
+                str(exc),
             )
 
-            deadline = time.monotonic() + self.log_timeout_seconds
+        return DDMCommandResult(
-            while time.monotonic() < deadline:
+            True,
-                try:
+            f"Alienware {self.alienware_model_token} was instructed to switch to {target_input} via monitorcontrol.",
-                    contents = log_path.read_text(encoding="utf-8", errors="replace")
+        )
-                except OSError:
-                    contents = ""
-                if contents.strip():
-                    return contents
-                time.sleep(0.1)
 
-            try:
+    def _resolve_alienware_monitor(self) -> tuple[object | None, str | None]:
-                return log_path.read_text(encoding="utf-8", errors="replace")
+        try:
-            except OSError:
+            monitors = list(get_monitors())
-                return ""
+        except Exception as exc:
-        finally:
+            return None, f"monitorcontrol could not enumerate monitors: {exc}"
-            try:
-                log_path.unlink()
-            except OSError:
-                pass
 
-    def _find_executable(self) -> Path | None:
+        candidates: list[object] = []
-        resolved = shutil.which("DDM.exe")
+        token = self.alienware_model_token.upper()
-        if resolved:
+        for monitor in monitors:
-            return Path(resolved)
+            description = str(getattr(getattr(monitor, "vcp", None), "description", "") or str(monitor))
-        if DEFAULT_DDM_PATH.exists():
+            if token in description.upper():
-            return DEFAULT_DDM_PATH
+                candidates.append(monitor)
-        return None
+
+        if not candidates:
+            return None, f"Alienware {self.alienware_model_token} was not found via monitorcontrol."
+        if len(candidates) > 1:
+            return None, f"Multiple Alienware {self.alienware_model_token} monitors were detected; refusing to switch."
+        return candidates[0], None
 
 
-def _extract_result_line(text: str, prefix: str) -> str:
+def _to_monitor_input_value(port_name: str) -> int | str:
-    for line in text.splitlines():
+    normalized = port_name.strip().upper()
-        normalized = line.strip()
+    if normalized == "DP2":
-        if normalized.startswith(prefix):
+        # Alienware AW3423DWF uses vendor-specific DP2 code.
-            return normalized
+        return 19
-    return ""
+    if normalized == "HDMI":
-
+        return "HDMI1"
-
+    return normalized
-def _extract_result_line_fragment(text: str, fragment: str) -> str:
-    needle = fragment.upper()
-    for line in text.splitlines():
-        normalized = line.strip()
-        if needle in normalized.upper():
-            return normalized
-    return ""
-
-
-def _output_has_failure(line: str, output: str) -> bool:
-    normalized_line = line.upper()
-    normalized_output = output.upper()
-    checks = (
-        "INVALID COMMAND",
-        "UNSUPPORTED",
-        "NOT FOUND",
-        "FAILED",
-        "FAIL",
-    )
-    return any(token in normalized_line or token in normalized_output for token in checks)

app/hardware.py

@@ -80,10 +80,6 @@ class RealMonitorBackend:
                 "Samsung trigger monitor",
                 errors,
             )
-            if not samsung_named_candidates:
-                errors.append(
-                    "Samsung trigger monitor did not expose a Samsung model in DDC/CI; using the only non-Alienware DDC monitor as trigger."
-                )
         elif not trigger_candidates:
             errors.append("Samsung trigger monitor could not be identified through DDC/CI.")
         else:

app/service.py

@@ -122,9 +122,9 @@ class KvmSwitcherService:
             ddm_slot is not None or supports_monitor_targeting
         )
         if not self.ddm_backend.is_available():
-            errors.append("DDM.exe was not found.")
+            errors.append("Alienware monitorcontrol backend is not ready.")
         elif ddm_slot is None and not supports_monitor_targeting:
-            errors.append("Alienware DDM slot could not be resolved.")
+            errors.append("Alienware monitorcontrol target could not be resolved.")
 
         trigger_target_port = self._port_name_for_input_code(scan.trigger_input_code)
         desired_port = self._resolve_desired_port(config, scan)
@@ -279,7 +279,7 @@ class KvmSwitcherService:
                 return "waiting_for_trigger_match", last_switch_at, scan, errors
             if verify_scan.alienware_input_code is None:
                 errors.append(
-                    "Alienware input could not be read after switch; assuming success because DDM command completed."
+                    "Alienware input could not be read after switch; assuming success because switch command completed."
                 )
                 self._samsung_session_successful = True
                 return "switched_unverified", last_switch_at, scan, errors
@@ -307,7 +307,7 @@ def _is_blocking_error(error: str) -> bool:
     normalized = error.strip()
     non_blocking_prefixes = (
         "Unable to read Alienware target monitor input source:",
-        "Alienware input could not be read after switch; assuming success because DDM command completed.",
+        "Alienware input could not be read after switch; assuming success because switch command completed.",
         "Samsung trigger monitor did not expose a Samsung model in DDC/CI; using the only non-Alienware DDC monitor as trigger.",
     )
     return not any(normalized.startswith(prefix) for prefix in non_blocking_prefixes)

static/index.html

@@ -323,14 +323,14 @@
         stat("Successful This Samsung Session", payload.samsung_session_successful ? "Yes" : "No"),
         stat("Attempts This Samsung Session", payload.samsung_session_attempt_count),
         stat("Waiting For Samsung Disconnect", payload.waiting_for_samsung_disconnect ? "Yes" : "No"),
-        stat("DDM Slot", payload.ddm_slot),
+        stat("Target Slot", payload.ddm_slot),
-        stat("DDM Ready", payload.ddm_ready ? "Yes" : "No"),
+        stat("Switch Ready", payload.ddm_ready ? "Yes" : "No"),
         stat("Last Result", payload.last_switch_result),
         stat("Last Switch At", payload.last_switch_at),
       ].join("");
 
       const tagHtml = [];
-      tagHtml.push(`<span class="tag${payload.ddm_ready ? "" : " warn"}">DDM ${payload.ddm_ready ? "Ready" : "Not Ready"}</span>`);
+      tagHtml.push(`<span class="tag${payload.ddm_ready ? "" : " warn"}">Switch ${payload.ddm_ready ? "Ready" : "Not Ready"}</span>`);
       tagHtml.push(`<span class="tag${payload.alienware_detected ? "" : " warn"}">Alienware ${payload.alienware_detected ? "Detected" : "Missing"}</span>`);
       tagHtml.push(`<span class="tag${payload.samsung_present ? "" : " warn"}">Samsung ${payload.samsung_present ? "Present" : "Missing"}</span>`);
       tagHtml.push(`<span class="tag${payload.samsung_session_successful ? "" : " warn"}">Session ${payload.samsung_session_successful ? "Successful" : "Pending"}</span>`);