Stabilize Wi-Fi by preventing AP fallback flapping

This commit is contained in:
2026-02-13 02:38:19 +01:00
parent 761fc49461
commit dcc3e2052c
2 changed files with 26 additions and 17 deletions

View File

@@ -22,6 +22,8 @@ class Supervisor(threading.Thread):
self.rtc = rtc
self._stop_event = threading.Event()
self._ntp_synced = False
self._wifi_down_streak = 0
self._ap_enable_after_cycles = 4 # 4 * 15s = 60s sustained Wi-Fi down
def stop(self) -> None:
self._stop_event.set()
@@ -32,17 +34,26 @@ class Supervisor(threading.Thread):
self.nm.refresh_state()
snapshot = self.state.snapshot()
should_have_ap = not (snapshot["wifi_connected"] and snapshot["internet_available"])
in_grace = time.time() < self.state.ap_grace_until
wifi_connected = bool(snapshot["wifi_connected"])
if should_have_ap and not snapshot["ap_mode"] and not snapshot["connecting"] and not in_grace:
LOG.warning("No active internet, enabling AP fallback")
if wifi_connected:
self._wifi_down_streak = 0
else:
self._wifi_down_streak += 1
if (
self._wifi_down_streak >= self._ap_enable_after_cycles
and not snapshot["ap_mode"]
and not snapshot["connecting"]
and not in_grace
):
LOG.warning(
"Wi-Fi disconnected for %ss, enabling AP fallback",
self._wifi_down_streak * 15,
)
self.nm.start_ap()
if not should_have_ap and snapshot["ap_mode"]:
LOG.info("Internet restored, disabling AP")
self.nm.stop_ap()
if snapshot["internet_available"] and not self._ntp_synced:
ok, msg = self.rtc.sync_ntp_and_rtc()
if ok:

View File

@@ -91,16 +91,14 @@ class NetworkManager:
return False
if not self.has_default_route():
return False
try:
with socket.create_connection(("1.1.1.1", 53), timeout=2):
pass
except OSError:
return False
try:
socket.gethostbyname("pool.ntp.org")
except OSError:
return False
return True
# Avoid DNS lookups here; transient resolver issues should not trigger AP mode flapping.
for target in (("1.1.1.1", 53), ("8.8.8.8", 53)):
try:
with socket.create_connection(target, timeout=2):
return True
except OSError:
continue
return False
def _service_action(self, action: str, unit: str) -> bool:
return self._run_quiet(["systemctl", action, unit], timeout=20)