Refactor KVM switch logic to enforce input source based on auxiliary monitor presence and update related documentation and tests

README.md

@@ -17,14 +17,13 @@ Configure only one value per machine:
 - this device Alienware target port: `DP1`, `DP2`, or `HDMI`
 - optional auxiliary trigger monitor: selected in UI; if not set, auto-detection is used
 
-Samsung trigger input codes mapped by port:
+Samsung trigger input codes (for diagnostics only):
 
 - `DP1` -> `15`
 - `DP2` -> `19` (or `16` on some setups)
 - `HDMI` -> `17`
 
-- 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.
+- If the auxiliary external monitor is present, the app checks Alienware current input and enforces the configured `device_port` via `monitorcontrol`.
 
 Retry behavior:
 

app/service.py

@@ -80,7 +80,6 @@ class KvmSwitcherService:
         self._samsung_session_attempted = False
         self._samsung_session_successful = False
         self._samsung_session_attempt_count = 0
-        self._last_trigger_input_code: int | None = None
         self._state_lock = Lock()
         self._stop_event = Event()
         self._thread: Thread | None = None
@@ -124,7 +123,7 @@ class KvmSwitcherService:
             scan=scan,
             errors=errors,
         )
-        self._update_samsung_session(scan.samsung_present, trigger_input_code)
+        self._update_samsung_session(scan.samsung_present)
 
         ddm_slot = self.ddm_backend.resolve_alienware_slot(force=False)
         supports_monitor_targeting = bool(
@@ -140,8 +139,8 @@ class KvmSwitcherService:
         blocking_errors = [error for error in errors if _is_blocking_error(error)]
 
         trigger_target_port = self._port_name_for_input_code(trigger_input_code)
-        desired_port = self._resolve_desired_port(config, trigger_input_code)
-        trigger_matches_device_port = desired_port is not None
+        desired_port = self._resolve_desired_port(config)
+        trigger_matches_device_port = scan.samsung_present
         last_switch_result = "idle"
         with self._state_lock:
             last_switch_at = self._status.last_switch_at
@@ -159,8 +158,9 @@ class KvmSwitcherService:
             desired_codes = self._port_input_codes(desired_port)
             if scan.alienware_input_code is not None and scan.alienware_input_code in desired_codes:
                 last_switch_result = "noop"
-                self._samsung_session_attempted = True
+                self._samsung_session_attempted = False
                 self._samsung_session_successful = True
+                self._samsung_session_attempt_count = 0
             else:
                 last_switch_result, last_switch_at, scan, errors = self._attempt_switch_sequence(
                     config=config,
@@ -170,12 +170,10 @@ class KvmSwitcherService:
                     last_switch_at=last_switch_at,
                     errors=errors,
                 )
-        elif trigger_matches_device_port and self._samsung_session_successful:
+        elif scan.samsung_present and self._samsung_session_successful:
             last_switch_result = "waiting_for_disconnect"
-        elif trigger_matches_device_port and self._samsung_session_attempt_count >= 3:
+        elif scan.samsung_present and self._samsung_session_attempt_count >= 3:
             last_switch_result = "max_attempts_waiting_for_disconnect"
-        elif scan.samsung_present:
-            last_switch_result = "waiting_for_trigger_match"
 
         status = ServiceStatus(
             config=config,
@@ -217,14 +215,8 @@ class KvmSwitcherService:
     @staticmethod
     def _resolve_desired_port(
         config: AppConfig,
-        trigger_input: int | None,
     ) -> str | None:
-        if trigger_input is None:
-            return None
-        desired_codes = KvmSwitcherService._port_input_codes(config.device_port)
-        if trigger_input in desired_codes:
         return config.device_port
-        return None
 
     @staticmethod
     def _port_input_codes(port_name: str) -> set[int]:
@@ -243,13 +235,12 @@ class KvmSwitcherService:
                 return port_name
         return None
 
-    def _update_samsung_session(self, samsung_present: bool, trigger_input_code: int | None) -> None:
+    def _update_samsung_session(self, samsung_present: bool) -> None:
         if not samsung_present:
             self._samsung_session_active = False
             self._samsung_session_attempted = False
             self._samsung_session_successful = False
             self._samsung_session_attempt_count = 0
-            self._last_trigger_input_code = None
             return
 
         if not self._samsung_session_active:
@@ -257,14 +248,6 @@ class KvmSwitcherService:
             self._samsung_session_attempted = False
             self._samsung_session_successful = False
             self._samsung_session_attempt_count = 0
-            self._last_trigger_input_code = trigger_input_code
-            return
-
-        if trigger_input_code != self._last_trigger_input_code:
-            self._samsung_session_attempted = False
-            self._samsung_session_successful = False
-            self._samsung_session_attempt_count = 0
-            self._last_trigger_input_code = trigger_input_code
 
     def _attempt_switch_sequence(
         self,
@@ -295,28 +278,28 @@ class KvmSwitcherService:
             verify_scan = self.monitor_backend.scan()
             scan = verify_scan
             errors.extend(verify_scan.errors)
-            verify_trigger_input, _, _ = self._resolve_trigger_state(
+            self._resolve_trigger_state(
                 config=config,
                 scan=verify_scan,
                 errors=errors,
             )
-            self._update_samsung_session(verify_scan.samsung_present, verify_trigger_input)
+            self._update_samsung_session(verify_scan.samsung_present)
 
             if not verify_scan.samsung_present:
                 return "waiting_for_reconnect", last_switch_at, scan, errors
 
             desired_codes = self._port_input_codes(desired_port)
-            if verify_trigger_input not in desired_codes:
-                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 switch command completed."
                 )
                 self._samsung_session_successful = True
+                self._samsung_session_attempt_count = 0
                 return "switched_unverified", last_switch_at, scan, errors
 
             if verify_scan.alienware_input_code in desired_codes:
                 self._samsung_session_successful = True
+                self._samsung_session_attempt_count = 0
                 return last_result, last_switch_at, scan, errors
 
         return "max_attempts_waiting_for_disconnect", last_switch_at, scan, errors

static/index.html

@@ -250,7 +250,7 @@
       <p class="eyebrow">Alienware-Only Targeting</p>
       <h1>Internal KVM Switch Dashboard</h1>
       <p>
-        Samsung is trigger-only. This device switches Alienware AW3423DWF when Samsung input matches the configured local port code: DP1 = 15, DP2 = 19 (or 16 on some setups), HDMI = 17.
+        When the auxiliary external monitor is present, this device enforces the configured Alienware AW3423DWF input (DP1, DP2, or HDMI).
       </p>
     </section>
 
@@ -362,7 +362,7 @@
         stat("Samsung Present", payload.samsung_present ? "Yes" : "No"),
         stat("Trigger Input", payload.trigger_input_code),
         stat("Trigger Target Port", payload.trigger_target_port),
-        stat("Trigger Matches This Device Port", payload.trigger_matches_device_port ? "Yes" : "No"),
+        stat("External Monitor Triggered", payload.trigger_matches_device_port ? "Yes" : "No"),
         stat("Alienware Detected", payload.alienware_detected ? "Yes" : "No"),
         stat("Alienware Input", payload.alienware_input_code),
         stat("Attempted This Samsung Session", payload.samsung_session_attempted ? "Yes" : "No"),

tests/test_app.py

@@ -266,7 +266,7 @@ def test_polling_uses_configured_auxiliary_monitor() -> None:
     assert not status["errors"]
 
 
-def test_polling_does_not_switch_when_trigger_does_not_match_device_port() -> None:
+def test_polling_switches_even_when_trigger_input_does_not_match_device_port() -> None:
     service_module = _require_module("app.service")
     hardware_module = _require_module("app.hardware")
     ddm_module = _require_module("app.ddm")
@@ -278,12 +278,17 @@ def test_polling_does_not_switch_when_trigger_does_not_match_device_port() -> No
         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
+            alienware_input = 15 if self.call_count == 1 else 19
             return HardwareScan(
                 samsung_present=True,
                 trigger_input_code=15,
                 alienware_detected=True,
-                alienware_input_code=15,
+                alienware_input_code=alienware_input,
                 errors=[],
             )
 
@@ -314,9 +319,9 @@ def test_polling_does_not_switch_when_trigger_does_not_match_device_port() -> No
         retry_wait_seconds=0.0,
     )
     status = service.save_settings(device_port="DP2")
-    assert status["trigger_matches_device_port"] is False
-    assert status["last_switch_result"] == "waiting_for_trigger_match"
-    assert backend.calls == []
+    assert status["trigger_matches_device_port"] is True
+    assert status["last_switch_result"] == "switched"
+    assert backend.calls == [(1, "DP2")]
 
 
 def test_polling_switches_with_monitor_targeting_when_slot_is_unavailable() -> None:
@@ -443,7 +448,7 @@ def test_polling_switches_when_alienware_input_is_unreadable() -> None:
     assert backend.calls == [(1, "DP1")]
 
 
-def test_polling_resets_session_when_trigger_changes_without_disconnect() -> None:
+def test_polling_does_not_reswitch_after_manual_change_while_monitor_stays_connected() -> None:
     service_module = _require_module("app.service")
     hardware_module = _require_module("app.hardware")
     ddm_module = _require_module("app.ddm")
@@ -521,7 +526,89 @@ def test_polling_resets_session_when_trigger_changes_without_disconnect() -> Non
     first = service.save_settings(device_port="DP2")
     assert first["last_switch_result"] == "noop"
     second = service.poll_once()
-    assert second["last_switch_result"] == "waiting_for_trigger_match"
+    assert second["last_switch_result"] == "waiting_for_disconnect"
+    assert backend.calls == []
+
+
+def test_polling_switches_again_after_aux_monitor_disconnect_and_reconnect() -> None:
+    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=19,
+                    alienware_detected=True,
+                    alienware_input_code=19,
+                    errors=[],
+                )
+            if self.call_count == 2:
+                return HardwareScan(
+                    samsung_present=False,
+                    trigger_input_code=None,
+                    alienware_detected=True,
+                    alienware_input_code=15,
+                    errors=[],
+                )
+            if self.call_count == 3:
+                return HardwareScan(
+                    samsung_present=True,
+                    trigger_input_code=19,
+                    alienware_detected=True,
+                    alienware_input_code=15,
+                    errors=[],
+                )
+            return HardwareScan(
+                samsung_present=True,
+                trigger_input_code=19,
+                alienware_detected=True,
+                alienware_input_code=19,
+                errors=[],
+            )
+
+    class FakeDDMBackend:
+        def __init__(self) -> None:
+            self.calls: list[tuple[int, str]] = []
+
+        def is_available(self) -> bool:
+            return True
+
+        def resolve_alienware_slot(self, force: bool = False) -> int | None:
+            return 1
+
+        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,
+    )
+    first = service.save_settings(device_port="DP2")
+    assert first["last_switch_result"] == "noop"
+    second = service.poll_once()
+    assert second["last_switch_result"] == "idle"
     third = service.poll_once()
     assert third["last_switch_result"] == "switched"
     assert backend.calls == [(1, "DP2")]