diff --git a/README.md b/README.md index e83175e..cc88dae 100644 --- a/README.md +++ b/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 `. +- If Samsung trigger input matches this device port code, the app runs `monitor.set_input_source()` on `AW3423DWF` via `monitorcontrol`. - If it does not match, the app waits. Retry behavior: diff --git a/app/ddm.py b/app/ddm.py index e20ca51..4e5d7a6 100644 --- a/app/ddm.py +++ b/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, - max_slots: int = 8, - command_timeout_seconds: float = 4.0, - log_timeout_seconds: float = 3.0, + alienware_model_token: str = ALIENWARE_MODEL_TOKEN, ): - self.executable_path = executable_path or self._find_executable() - self.max_slots = max_slots - self.command_timeout_seconds = command_timeout_seconds - self.log_timeout_seconds = log_timeout_seconds + self.alienware_model_token = alienware_model_token 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): - result = self._run_logged_command(f"/{slot}:ReadAssetAttributes") - line = _extract_result_line(result, f"{slot}:ReadAssetAttributes") - if not line: - continue - if "INVALID COMMAND" in line.upper(): - continue - if ALIENWARE_MODEL_TOKEN in line.upper(): - self._cached_slot = slot - return slot - - self._cached_slot = None - return None + monitor, error = self._resolve_alienware_monitor() + del monitor + if error is not None: + self._cached_slot = None + return None + # Service still expects a slot-like value. With monitorcontrol we + # target by model, so we expose a synthetic stable slot. + self._cached_slot = 1 + return self._cached_slot def switch_to_port(self, slot: int, port_name: str) -> DDMCommandResult: - if not self.is_available(): - return DDMCommandResult(False, "DDM.exe was not found.") - + del slot 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"]) - monitor_selector = f"/MNT:{ALIENWARE_MODEL_TOKEN}" - - targeted_output = self._run_logged_command( - monitor_selector, - "/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) + target_input = _to_monitor_input_value(str(port_spec["ddm_input"])) + monitor, error = self._resolve_alienware_monitor() + if error is not None: + return DDMCommandResult(False, error) + assert monitor is not None try: - subprocess.run( - [str(self.executable_path), "/Log", str(log_path), *command_args], - check=False, - capture_output=True, - text=True, - timeout=self.command_timeout_seconds, + with monitor: + monitor.set_input_source(target_input) + except Exception as exc: + return DDMCommandResult( + False, + f"monitorcontrol failed to switch Alienware input to {target_input}: {exc}", + str(exc), ) - deadline = time.monotonic() + self.log_timeout_seconds - while time.monotonic() < deadline: - try: - contents = log_path.read_text(encoding="utf-8", errors="replace") - except OSError: - contents = "" - if contents.strip(): - return contents - time.sleep(0.1) + return DDMCommandResult( + True, + f"Alienware {self.alienware_model_token} was instructed to switch to {target_input} via monitorcontrol.", + ) - try: - return log_path.read_text(encoding="utf-8", errors="replace") - except OSError: - return "" - finally: - try: - log_path.unlink() - except OSError: - pass + def _resolve_alienware_monitor(self) -> tuple[object | None, str | None]: + try: + monitors = list(get_monitors()) + except Exception as exc: + return None, f"monitorcontrol could not enumerate monitors: {exc}" - def _find_executable(self) -> Path | None: - resolved = shutil.which("DDM.exe") - if resolved: - return Path(resolved) - if DEFAULT_DDM_PATH.exists(): - return DEFAULT_DDM_PATH - return None + candidates: list[object] = [] + token = self.alienware_model_token.upper() + for monitor in monitors: + description = str(getattr(getattr(monitor, "vcp", None), "description", "") or str(monitor)) + if token in description.upper(): + candidates.append(monitor) + + 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: - for line in text.splitlines(): - normalized = line.strip() - if normalized.startswith(prefix): - return normalized - return "" - - -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) +def _to_monitor_input_value(port_name: str) -> int | str: + normalized = port_name.strip().upper() + if normalized == "DP2": + # Alienware AW3423DWF uses vendor-specific DP2 code. + return 19 + if normalized == "HDMI": + return "HDMI1" + return normalized diff --git a/app/hardware.py b/app/hardware.py index 12e51bc..aa8a50c 100644 --- a/app/hardware.py +++ b/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: diff --git a/app/service.py b/app/service.py index 8723bc3..f6e31db 100644 --- a/app/service.py +++ b/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) diff --git a/static/index.html b/static/index.html index bb6d84e..497d69d 100644 --- a/static/index.html +++ b/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("DDM Ready", payload.ddm_ready ? "Yes" : "No"), + stat("Target Slot", payload.ddm_slot), + 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(`DDM ${payload.ddm_ready ? "Ready" : "Not Ready"}`); + tagHtml.push(`Switch ${payload.ddm_ready ? "Ready" : "Not Ready"}`); tagHtml.push(`Alienware ${payload.alienware_detected ? "Detected" : "Missing"}`); tagHtml.push(`Samsung ${payload.samsung_present ? "Present" : "Missing"}`); tagHtml.push(`Session ${payload.samsung_session_successful ? "Successful" : "Pending"}`);