From 2416e8bf0a68e23cb207db2fa0474c303d809561 Mon Sep 17 00:00:00 2001 From: Chuck Date: Mon, 13 Jul 2026 07:40:58 -0400 Subject: [PATCH 1/2] perf(wifi): one status fetch per monitor tick instead of three MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wifi monitor daemon fetched WiFi status + ethernet state before its AP-mode check, the check internally fetched the same state again (with retry), and the daemon fetched a third time afterwards — each fetch is several nmcli subprocess forks, every 30s, forever, even on a perfectly healthy link. check_and_manage_ap_mode's decision logic is extracted to _manage_ap_mode(status, ethernet, ap_active); the new check_and_manage_ap_mode_with_state() runs the single (retrying) fetch battery and returns (changed, status, ethernet, ap_active_after) — the post-state is derivable because state only ever flips via one enable or one disable. The original bool-returning method delegates, so existing callers are untouched. The daemon's dead pre-fetch is removed and its post-check reads use the returned state; the AP-enable retry semantics (_get_wifi_status_with_retry) are preserved exactly. ~10-15 forks/30s drops to ~4-6. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam --- scripts/utils/wifi_monitor_daemon.py | 28 ++++----- src/wifi_manager.py | 24 ++++++++ test/test_wifi_check_state.py | 92 ++++++++++++++++++++++++++++ 3 files changed, 128 insertions(+), 16 deletions(-) create mode 100644 test/test_wifi_check_state.py diff --git a/scripts/utils/wifi_monitor_daemon.py b/scripts/utils/wifi_monitor_daemon.py index 53f0497cb..414c16dc4 100755 --- a/scripts/utils/wifi_monitor_daemon.py +++ b/scripts/utils/wifi_monitor_daemon.py @@ -78,21 +78,17 @@ def run(self): while self.running: try: - # Get current status before checking - status = self.wifi_manager.get_wifi_status() - ethernet_connected = self.wifi_manager._is_ethernet_connected() - - # Check WiFi status and manage AP mode - state_changed = self.wifi_manager.check_and_manage_ap_mode() - - # Get updated status after check - updated_status = self.wifi_manager.get_wifi_status() - updated_ethernet = self.wifi_manager._is_ethernet_connected() - + # One combined check that also returns the state it observed — + # the previous flow fetched status before AND after the check + # on top of the check's own internal fetch, each one several + # nmcli subprocess forks, every 30s, forever. + (state_changed, updated_status, updated_ethernet, + ap_active) = self.wifi_manager.check_and_manage_ap_mode_with_state() + current_state = { 'connected': updated_status.connected, 'ethernet_connected': updated_ethernet, - 'ap_active': updated_status.ap_mode_active, + 'ap_active': ap_active, 'ssid': updated_status.ssid } @@ -109,7 +105,7 @@ def run(self): else: logger.debug("Ethernet not connected") - if updated_status.ap_mode_active: + if ap_active: logger.info(f"AP mode ACTIVE - SSID: {ap_ssid} (IP: 192.168.4.1)") else: logger.debug("AP mode inactive") @@ -123,16 +119,16 @@ def run(self): # Log periodic status (less verbose) if updated_status.connected: logger.debug(f"Status check: WiFi={updated_status.ssid} ({updated_status.signal}%), " - f"Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}") + f"Ethernet={updated_ethernet}, AP={ap_active}") else: - logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}") + logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={ap_active}") # Escalating recovery: if nmcli reports connected but actual internet # is unreachable for several consecutive checks, restart NetworkManager. # This is done HERE (not inside check_and_manage_ap_mode) to keep the # AP-enable trigger clean and avoid false-positive AP enables from # transient packet loss on otherwise working WiFi. - if updated_status.connected and not updated_status.ap_mode_active: + if updated_status.connected and not ap_active: if not self.wifi_manager.check_internet_connectivity(): self._consecutive_internet_failures += 1 logger.warning( diff --git a/src/wifi_manager.py b/src/wifi_manager.py index 0983ff3f8..5d33ecd94 100644 --- a/src/wifi_manager.py +++ b/src/wifi_manager.py @@ -2564,11 +2564,35 @@ def check_and_manage_ap_mode(self) -> bool: Returns: True if AP mode state changed, False otherwise """ + changed, _status, _ethernet, _ap = self.check_and_manage_ap_mode_with_state() + return changed + + def check_and_manage_ap_mode_with_state(self): + """Like check_and_manage_ap_mode, but also returns the state it + observed, so callers (the wifi monitor daemon) don't have to re-run + the same nmcli subprocess battery before AND after the check — + each status fetch is several process forks. + + Returns: + (state_changed, WiFiStatus, ethernet_connected, ap_active_after) + """ try: # Get status with retry for more reliable detection status = self._get_wifi_status_with_retry() ethernet_connected = self._is_ethernet_connected() ap_active = self._is_ap_mode_active() + changed = self._manage_ap_mode(status, ethernet_connected, ap_active) + # State only ever changes via one enable or one disable, so the + # post-state is the inverse of the pre-state when changed. + ap_after = (not ap_active) if changed else ap_active + return changed, status, ethernet_connected, ap_after + except Exception as e: + logger.error(f"Error checking AP mode: {e}", exc_info=True) + return False, WiFiStatus(connected=False), False, False + + def _manage_ap_mode(self, status, ethernet_connected, ap_active) -> bool: + """AP-mode decision logic against an already-fetched state snapshot.""" + try: auto_enable = self.config.get("auto_enable_ap_mode", True) # Default: True (safe due to grace period) # Log current state for debugging diff --git a/test/test_wifi_check_state.py b/test/test_wifi_check_state.py new file mode 100644 index 000000000..cb5c268c7 --- /dev/null +++ b/test/test_wifi_check_state.py @@ -0,0 +1,92 @@ +"""Tests for check_and_manage_ap_mode_with_state (src/wifi_manager.py). + +The wifi monitor daemon used to fetch WiFi status before AND after each +check on top of the check's own internal fetch — every fetch is several +nmcli subprocess forks, every 30s, forever. The new API returns the state +the check observed, so the daemon runs exactly one fetch battery per tick. +""" + +import os +import sys +from unittest.mock import MagicMock, patch + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from src.wifi_manager import WiFiManager, WiFiStatus # noqa: E402 + + +@pytest.fixture +def wm(tmp_path): + with patch.object(WiFiManager, "_load_config", return_value={}, create=True): + manager = WiFiManager.__new__(WiFiManager) + # minimal attribute setup without running the real __init__ + manager.config = {"auto_enable_ap_mode": True} + manager._disconnected_checks = 0 + manager._disconnected_checks_required = 3 + manager._ap_enabled_at = None + return manager + + +def _wire(wm, connected, ethernet, ap_active): + wm._get_wifi_status_with_retry = MagicMock( + return_value=WiFiStatus(connected=connected, ssid="net" if connected else None)) + wm._is_ethernet_connected = MagicMock(return_value=ethernet) + wm._is_ap_mode_active = MagicMock(return_value=ap_active) + wm.enable_ap_mode = MagicMock(return_value=(True, "ok")) + wm.disable_ap_mode = MagicMock(return_value=(True, "ok")) + wm.scan_networks = MagicMock(return_value=([], False)) + wm._save_cached_scan = MagicMock() + wm._FORCE_AP_FLAG_PATH = MagicMock() + wm._FORCE_AP_FLAG_PATH.exists.return_value = False + + +class TestWithState: + def test_single_fetch_per_call(self, wm): + _wire(wm, connected=True, ethernet=False, ap_active=False) + wm.check_and_manage_ap_mode_with_state() + assert wm._get_wifi_status_with_retry.call_count == 1 + assert wm._is_ethernet_connected.call_count == 1 + assert wm._is_ap_mode_active.call_count == 1 + + def test_returns_observed_state(self, wm): + _wire(wm, connected=True, ethernet=True, ap_active=False) + changed, status, ethernet, ap_after = wm.check_and_manage_ap_mode_with_state() + assert changed is False + assert status.connected is True + assert ethernet is True + assert ap_after is False + + def test_ap_after_inverts_on_disable(self, wm): + """WiFi reconnects while AP is up -> auto-disable -> ap_after False.""" + _wire(wm, connected=True, ethernet=False, ap_active=True) + changed, status, ethernet, ap_after = wm.check_and_manage_ap_mode_with_state() + assert changed is True + assert ap_after is False + wm.disable_ap_mode.assert_called_once() + + def test_ap_after_inverts_on_enable(self, wm): + """Grace period exhausted with nothing connected -> enable -> True.""" + _wire(wm, connected=False, ethernet=False, ap_active=False) + wm._disconnected_checks = 2 # this call is the 3rd + changed, status, ethernet, ap_after = wm.check_and_manage_ap_mode_with_state() + assert changed is True + assert ap_after is True + wm.enable_ap_mode.assert_called_once() + + def test_bool_wrapper_is_back_compatible(self, wm): + _wire(wm, connected=True, ethernet=False, ap_active=False) + assert wm.check_and_manage_ap_mode() is False + _wire(wm, connected=True, ethernet=False, ap_active=True) + assert wm.check_and_manage_ap_mode() is True + + def test_exception_path_never_raises(self, wm): + wm._get_wifi_status_with_retry = MagicMock(side_effect=RuntimeError("nmcli gone")) + changed, status, ethernet, ap_after = wm.check_and_manage_ap_mode_with_state() + assert changed is False + assert status.connected is False + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-v"])) From a26b777fbe965d29f99998bb267c27582d6cb184 Mon Sep 17 00:00:00 2001 From: ChuckBuilds Date: Tue, 14 Jul 2026 11:08:14 -0400 Subject: [PATCH 2/2] fix(wifi): add missing type hints on AP-mode state helpers, fix unused-var lint in tests - check_and_manage_ap_mode_with_state now declares its Tuple[bool, WiFiStatus, bool, bool] return type instead of being untyped. - _manage_ap_mode's status/ethernet_connected/ap_active parameters are now typed (WiFiStatus, bool, bool), matching its existing -> bool return hint. - test_wifi_check_state.py: rename unused unpacked variables (status/ ethernet/ap_after) to underscore-prefixed names at the three call sites that don't use them, leaving the used ones untouched. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ --- src/wifi_manager.py | 4 ++-- test/test_wifi_check_state.py | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/wifi_manager.py b/src/wifi_manager.py index 5d33ecd94..808767ff3 100644 --- a/src/wifi_manager.py +++ b/src/wifi_manager.py @@ -2567,7 +2567,7 @@ def check_and_manage_ap_mode(self) -> bool: changed, _status, _ethernet, _ap = self.check_and_manage_ap_mode_with_state() return changed - def check_and_manage_ap_mode_with_state(self): + def check_and_manage_ap_mode_with_state(self) -> Tuple[bool, WiFiStatus, bool, bool]: """Like check_and_manage_ap_mode, but also returns the state it observed, so callers (the wifi monitor daemon) don't have to re-run the same nmcli subprocess battery before AND after the check — @@ -2590,7 +2590,7 @@ def check_and_manage_ap_mode_with_state(self): logger.error(f"Error checking AP mode: {e}", exc_info=True) return False, WiFiStatus(connected=False), False, False - def _manage_ap_mode(self, status, ethernet_connected, ap_active) -> bool: + def _manage_ap_mode(self, status: WiFiStatus, ethernet_connected: bool, ap_active: bool) -> bool: """AP-mode decision logic against an already-fetched state snapshot.""" try: auto_enable = self.config.get("auto_enable_ap_mode", True) # Default: True (safe due to grace period) diff --git a/test/test_wifi_check_state.py b/test/test_wifi_check_state.py index cb5c268c7..349e80a05 100644 --- a/test/test_wifi_check_state.py +++ b/test/test_wifi_check_state.py @@ -61,7 +61,7 @@ def test_returns_observed_state(self, wm): def test_ap_after_inverts_on_disable(self, wm): """WiFi reconnects while AP is up -> auto-disable -> ap_after False.""" _wire(wm, connected=True, ethernet=False, ap_active=True) - changed, status, ethernet, ap_after = wm.check_and_manage_ap_mode_with_state() + changed, _status, _ethernet, ap_after = wm.check_and_manage_ap_mode_with_state() assert changed is True assert ap_after is False wm.disable_ap_mode.assert_called_once() @@ -70,7 +70,7 @@ def test_ap_after_inverts_on_enable(self, wm): """Grace period exhausted with nothing connected -> enable -> True.""" _wire(wm, connected=False, ethernet=False, ap_active=False) wm._disconnected_checks = 2 # this call is the 3rd - changed, status, ethernet, ap_after = wm.check_and_manage_ap_mode_with_state() + changed, _status, _ethernet, ap_after = wm.check_and_manage_ap_mode_with_state() assert changed is True assert ap_after is True wm.enable_ap_mode.assert_called_once() @@ -83,7 +83,7 @@ def test_bool_wrapper_is_back_compatible(self, wm): def test_exception_path_never_raises(self, wm): wm._get_wifi_status_with_retry = MagicMock(side_effect=RuntimeError("nmcli gone")) - changed, status, ethernet, ap_after = wm.check_and_manage_ap_mode_with_state() + changed, status, _ethernet, _ap_after = wm.check_and_manage_ap_mode_with_state() assert changed is False assert status.connected is False