diff --git a/README.md b/README.md index 709f0b9..56c4e84 100644 --- a/README.md +++ b/README.md @@ -30,8 +30,8 @@ Valid values are: Examples: -- Tower role with UI port `DP1` -> on Samsung trigger `15`, send `DDM.exe /1:WriteActiveInput DP1` -- Laptop role with UI port `DP2` -> on Samsung trigger `19`, send `DDM.exe /1:WriteActiveInput DP2` +- Tower role with UI port `DP1` -> on Samsung trigger `15`, send `DDM.exe /MNT:AW3423DWF /WriteActiveInput DP1` +- Laptop role with UI port `DP2` -> on Samsung trigger `19`, send `DDM.exe /MNT:AW3423DWF /WriteActiveInput DP2` Retry behavior: @@ -92,10 +92,10 @@ shell:startup 3. Create a shortcut in that folder pointing to: ```text -C:\Users\LagoWorkStation\OneDrive\Documentos\BE-terna\Internal - KVM Switch\start_kvm_switch.cmd +C:\Users\LagoWorkStation\OneDrive\Documentos\BE-terna\Internal - KVM Switch\start_kvm_switch_background.vbs ``` -This will start the app when you log in. It opens a console window. +This starts the app at logon in the background (no persistent console window). ### Cleaner option: Task Scheduler @@ -104,13 +104,13 @@ Use Task Scheduler if you want it to start automatically at logon with better co Program/script: ```text -cmd.exe +wscript.exe ``` Arguments: ```text -/c "C:\Users\LagoWorkStation\OneDrive\Documentos\BE-terna\Internal - KVM Switch\start_kvm_switch.cmd" +"C:\Users\LagoWorkStation\OneDrive\Documentos\BE-terna\Internal - KVM Switch\start_kvm_switch_background.vbs" ``` -Set the trigger to `At log on`. +Set the trigger to `At log on`. Optional: enable `Hidden` in task settings. diff --git a/app/ddm.py b/app/ddm.py index 91ccf4c..e20ca51 100644 --- a/app/ddm.py +++ b/app/ddm.py @@ -26,6 +26,8 @@ class DDMCommandResult: class DDMBackend(Protocol): def is_available(self) -> bool: ... + def supports_monitor_targeting(self) -> bool: ... + def resolve_alienware_slot(self, force: bool = False) -> int | None: ... def invalidate_slot(self) -> None: ... @@ -51,6 +53,9 @@ class RealDDMBackend: def is_available(self) -> bool: return self.executable_path is not None and self.executable_path.exists() + def supports_monitor_targeting(self) -> bool: + return True + def invalidate_slot(self) -> None: with self._lock: self._cached_slot = None @@ -86,17 +91,43 @@ class RealDDMBackend: return DDMCommandResult(False, f"Port {port_name} is not supported.") ddm_input = str(port_spec["ddm_input"]) + monitor_selector = f"/MNT:{ALIENWARE_MODEL_TOKEN}" - output = self._run_logged_command(f"/{slot}:WriteActiveInput", ddm_input) - line = _extract_result_line(output, f"{slot}:WriteActiveInput") - if line and "INVALID COMMAND" in line.upper(): - return DDMCommandResult(False, "DDM rejected the WriteActiveInput command.", output) - - return DDMCommandResult( - True, - f"Alienware DDM slot {slot} was instructed to switch to {ddm_input}.", - output, + 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 @@ -148,3 +179,25 @@ def _extract_result_line(text: str, prefix: str) -> str: 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) diff --git a/app/service.py b/app/service.py index 0594375..6647159 100644 --- a/app/service.py +++ b/app/service.py @@ -117,10 +117,15 @@ class KvmSwitcherService: errors.extend(config.validate()) ddm_slot = self.ddm_backend.resolve_alienware_slot(force=False) - ddm_ready = self.ddm_backend.is_available() and ddm_slot is not None + supports_monitor_targeting = bool( + getattr(self.ddm_backend, "supports_monitor_targeting", lambda: False)() + ) + ddm_ready = self.ddm_backend.is_available() and ( + ddm_slot is not None or supports_monitor_targeting + ) if not self.ddm_backend.is_available(): errors.append("DDM.exe was not found.") - elif ddm_slot is None: + elif ddm_slot is None and not supports_monitor_targeting: errors.append("Alienware DDM slot could not be resolved.") resolved_target, desired_port = self._resolve_target(config, scan) @@ -132,7 +137,7 @@ class KvmSwitcherService: not errors and desired_port is not None and scan.alienware_input_code is not None - and ddm_slot is not None + and ddm_ready and scan.samsung_present and not self._samsung_session_successful and self._samsung_session_attempt_count < 3 @@ -229,7 +234,7 @@ class KvmSwitcherService: def _attempt_switch_sequence( self, - ddm_slot: int, + ddm_slot: int | None, desired_port: str, scan: HardwareScan, last_switch_at: str | None, @@ -241,7 +246,8 @@ class KvmSwitcherService: self._samsung_session_attempted = True self._samsung_session_attempt_count += 1 - result = self.ddm_backend.switch_to_port(ddm_slot, desired_port) + switch_slot = ddm_slot if ddm_slot is not None else 0 + result = self.ddm_backend.switch_to_port(switch_slot, desired_port) if not result.success: errors.append(result.message) self.ddm_backend.invalidate_slot() diff --git a/start_kvm_switch_background.vbs b/start_kvm_switch_background.vbs new file mode 100644 index 0000000..3c95851 --- /dev/null +++ b/start_kvm_switch_background.vbs @@ -0,0 +1,12 @@ +Option Explicit + +Dim shell +Dim projectPath +Dim command + +Set shell = CreateObject("WScript.Shell") +projectPath = CreateObject("Scripting.FileSystemObject").GetParentFolderName(WScript.ScriptFullName) +command = "cmd.exe /c cd /d """ & projectPath & """ && uv run kvm-switch" + +' Run hidden and do not wait. +shell.Run command, 0, False diff --git a/tests/test_app.py b/tests/test_app.py index 44154a8..13ce164 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -167,3 +167,75 @@ def test_polling_switches_when_trigger_matches_device_role(monkeypatch: pytest.M assert status["samsung_session_attempted"] is True assert status["samsung_session_successful"] is True assert status["last_switch_result"] == "switched" + + +def test_polling_switches_with_monitor_targeting_when_slot_is_unavailable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv(DEVICE_ROLE_ENV_VAR, "tower") + service_module = _require_module("app.service") + hardware_module = _require_module("app.hardware") + ddm_module = _require_module("app.ddm") + + KvmSwitcherService = getattr(service_module, "KvmSwitcherService", None) + HardwareScan = getattr(hardware_module, "HardwareScan", None) + DDMCommandResult = getattr(ddm_module, "DDMCommandResult", None) + if KvmSwitcherService is None or HardwareScan is None or DDMCommandResult is None: + pytest.skip("Service backend interfaces are not available yet.") + + class FakeMonitorBackend: + def __init__(self) -> None: + self.call_count = 0 + + def scan(self): + self.call_count += 1 + if self.call_count == 1: + return HardwareScan( + samsung_present=True, + trigger_input_code=15, + alienware_detected=True, + alienware_input_code=19, + errors=[], + ) + return HardwareScan( + samsung_present=True, + trigger_input_code=15, + alienware_detected=True, + alienware_input_code=15, + errors=[], + ) + + class FakeDDMBackend: + def __init__(self) -> None: + self.calls: list[tuple[int, str]] = [] + + def is_available(self) -> bool: + return True + + def supports_monitor_targeting(self) -> bool: + return True + + def resolve_alienware_slot(self, force: bool = False) -> int | None: + return None + + def invalidate_slot(self) -> None: + return None + + def switch_to_port(self, slot: int, port_name: str): + self.calls.append((slot, port_name)) + return DDMCommandResult(True, "ok") + + config_path = Path(tempfile.mkdtemp()) / "config.json" + backend = FakeDDMBackend() + service = KvmSwitcherService( + config_store=ConfigStore(config_path), + monitor_backend=FakeMonitorBackend(), + ddm_backend=backend, + poll_interval_seconds=0.01, + retry_wait_seconds=0.0, + ) + status = service.save_settings(device_role="tower", device_port="DP1") + assert status["ddm_ready"] is True + assert status["ddm_slot"] is None + assert status["last_switch_result"] == "switched" + assert backend.calls == [(0, "DP1")]