perf(wifi): one status fetch per monitor tick instead of three (#411)

* perf(wifi): one status fetch per monitor tick instead of three

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam

* 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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ

---------

Co-authored-by: Chuck <chuck@example.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-07-14 11:22:37 -04:00
committed by GitHub
co-authored by Claude Sonnet 5 Chuck
parent bff13129c4
commit 14a59c863c
3 changed files with 128 additions and 16 deletions
+11 -15
View File
@@ -78,21 +78,17 @@ class WiFiMonitorDaemon:
while self.running: while self.running:
try: try:
# Get current status before checking # One combined check that also returns the state it observed —
status = self.wifi_manager.get_wifi_status() # the previous flow fetched status before AND after the check
ethernet_connected = self.wifi_manager._is_ethernet_connected() # on top of the check's own internal fetch, each one several
# nmcli subprocess forks, every 30s, forever.
# Check WiFi status and manage AP mode (state_changed, updated_status, updated_ethernet,
state_changed = self.wifi_manager.check_and_manage_ap_mode() ap_active) = self.wifi_manager.check_and_manage_ap_mode_with_state()
# Get updated status after check
updated_status = self.wifi_manager.get_wifi_status()
updated_ethernet = self.wifi_manager._is_ethernet_connected()
current_state = { current_state = {
'connected': updated_status.connected, 'connected': updated_status.connected,
'ethernet_connected': updated_ethernet, 'ethernet_connected': updated_ethernet,
'ap_active': updated_status.ap_mode_active, 'ap_active': ap_active,
'ssid': updated_status.ssid 'ssid': updated_status.ssid
} }
@@ -109,7 +105,7 @@ class WiFiMonitorDaemon:
else: else:
logger.debug("Ethernet not connected") 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)") logger.info(f"AP mode ACTIVE - SSID: {ap_ssid} (IP: 192.168.4.1)")
else: else:
logger.debug("AP mode inactive") logger.debug("AP mode inactive")
@@ -123,16 +119,16 @@ class WiFiMonitorDaemon:
# Log periodic status (less verbose) # Log periodic status (less verbose)
if updated_status.connected: if updated_status.connected:
logger.debug(f"Status check: WiFi={updated_status.ssid} ({updated_status.signal}%), " 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: 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 # Escalating recovery: if nmcli reports connected but actual internet
# is unreachable for several consecutive checks, restart NetworkManager. # is unreachable for several consecutive checks, restart NetworkManager.
# This is done HERE (not inside check_and_manage_ap_mode) to keep the # 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 # AP-enable trigger clean and avoid false-positive AP enables from
# transient packet loss on otherwise working WiFi. # 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(): if not self.wifi_manager.check_internet_connectivity():
self._consecutive_internet_failures += 1 self._consecutive_internet_failures += 1
logger.warning( logger.warning(
+24
View File
@@ -2564,11 +2564,35 @@ address=/detectportal.firefox.com/192.168.4.1
Returns: Returns:
True if AP mode state changed, False otherwise 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) -> 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 —
each status fetch is several process forks.
Returns:
(state_changed, WiFiStatus, ethernet_connected, ap_active_after)
"""
try: try:
# Get status with retry for more reliable detection # Get status with retry for more reliable detection
status = self._get_wifi_status_with_retry() status = self._get_wifi_status_with_retry()
ethernet_connected = self._is_ethernet_connected() ethernet_connected = self._is_ethernet_connected()
ap_active = self._is_ap_mode_active() 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: 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) auto_enable = self.config.get("auto_enable_ap_mode", True) # Default: True (safe due to grace period)
# Log current state for debugging # Log current state for debugging
+92
View File
@@ -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"]))