diff --git a/README.md b/README.md index c6e61bd..cdeb988 100644 --- a/README.md +++ b/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()` 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: diff --git a/app/service.py b/app/service.py index f2e4306..de16f80 100644 --- a/app/service.py +++ b/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 + return config.device_port @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 diff --git a/static/index.html b/static/index.html index 71477e4..d519465 100644 --- a/static/index.html +++ b/static/index.html @@ -250,7 +250,7 @@

Alienware-Only Targeting

Internal KVM Switch Dashboard

- 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).

@@ -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"), diff --git a/tests/test_app.py b/tests/test_app.py index 58a7a85..42dbf16 100644 --- a/tests/test_app.py +++ b/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")]