Detect network applet state changes (#112)

Author: edumucelli

docking/applets/network/applet.py

@@ -17,6 +17,7 @@
 
 import time
 from collections.abc import Callable
+from dataclasses import replace
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -33,6 +34,7 @@
 from docking.applets.network.render import create_icon
 from docking.applets.network.state import (
     AvailableNetwork,
+    NetworkState,
     TrafficCounters,
     build_tooltip,
     compute_speeds,
@@ -83,15 +85,8 @@ def __init__(self, icon_size: int, config: Config | None = None) -> None:
         self._nm_handler_id: int = 0
         self._nm_state_handler_id: int = 0
 
-        # State
-        self._is_connected = False
-        self._is_wifi = False
-        self._ssid = ""
-        self._signal_strength = 0
-        self._iface = ""
-        self._ip_address = ""
-        self._rx_speed = 0.0
-        self._tx_speed = 0.0
+        # Visible state. Frozen so equality gates re-renders.
+        self._state = NetworkState()
         # "download", "upload", or "none"
         self._speed_overlay = "download"
 
@@ -111,11 +106,11 @@ def create_icon(self, size: int):
         """Load network icon with optional speed overlay."""
         return create_icon(
             size=size,
-            is_connected=self._is_connected,
-            is_wifi=self._is_wifi,
-            signal_strength=self._signal_strength,
-            rx_speed=self._rx_speed,
-            tx_speed=self._tx_speed,
+            is_connected=self._state.is_connected,
+            is_wifi=self._state.is_wifi,
+            signal_strength=self._state.signal_strength,
+            rx_speed=self._state.rx_speed,
+            tx_speed=self._state.tx_speed,
             speed_overlay=self._speed_overlay,
         )
 
@@ -124,34 +119,35 @@ def refresh_tooltip(self) -> None:
 
     def get_menu_items(self) -> list:
         """Show connection info and common network actions."""
+        state = self._state
         status: list[Gtk.MenuItem] = []
-        if self._ssid:
+        if state.ssid:
             status.append(
                 disabled_menu_item(
                     _("WiFi: {ssid} ({pct}%)").format(
-                        ssid=self._ssid, pct=self._signal_strength
+                        ssid=state.ssid, pct=state.signal_strength
                     ),
                     gtk=Gtk,
                 )
             )
-        elif self._is_connected:
+        elif state.is_connected:
             status.append(
                 disabled_menu_item(
-                    _("Ethernet: {iface}").format(iface=self._iface),
+                    _("Ethernet: {iface}").format(iface=state.iface),
                     gtk=Gtk,
                 )
             )
         else:
             status.append(disabled_menu_item(_("Not connected"), gtk=Gtk))
 
-        if self._ip_address:
+        if state.ip_address:
             status.append(
-                disabled_menu_item(_("IP: {ip}").format(ip=self._ip_address), gtk=Gtk)
+                disabled_menu_item(_("IP: {ip}").format(ip=state.ip_address), gtk=Gtk)
             )
 
-        if self._is_connected:
-            down = format_speed(bps=self._rx_speed)
-            up = format_speed(bps=self._tx_speed)
+        if state.is_connected:
+            down = format_speed(bps=state.rx_speed)
+            up = format_speed(bps=state.tx_speed)
             status.append(disabled_menu_item(f"\u2193 {down}  \u2191 {up}", gtk=Gtk))
 
         settings: list[Gtk.MenuItem] = []
@@ -246,8 +242,7 @@ def _set_networking_enabled(self, *, enabled: bool) -> None:
                 exc,
             )
             return
-        self._update_nm_state()
-        self.present()
+        self._refresh_state()
 
     def _set_wireless_enabled(self, *, enabled: bool) -> None:
         if self._nm_client is None:
@@ -261,8 +256,7 @@ def _set_wireless_enabled(self, *, enabled: bool) -> None:
                 exc,
             )
             return
-        self._update_nm_state()
-        self.present()
+        self._refresh_state()
 
     def _build_available_networks_submenu(self) -> Gtk.MenuItem:
         item = Gtk.MenuItem(label=_("Available Networks"))
@@ -506,7 +500,7 @@ def start(self, notify: Callable[[], None]) -> None:
                 "notify::state",
                 self._on_nm_changed,
             )
-            self._update_nm_state()
+            self._refresh_state()
         except GLib.Error:
             log.bind(action="connect_nm").warning(
                 "Could not connect to NetworkManager",
@@ -529,21 +523,21 @@ def stop(self) -> None:
         super().stop()
 
     def _on_nm_changed(self, *_args: object) -> None:
-        """NM active-connections changed: update state immediately."""
-        self._update_nm_state()
-        self.present()
-
-    def _update_nm_state(self) -> None:
-        """Read current connection info from NetworkManager."""
+        """NM active-connections changed: refresh and re-render on change."""
+        self._refresh_state()
+
+    def _refresh_state(self) -> None:
+        """Recompute the full visible state and present only on change."""
+        new_state = self._compute_nm_state()
+        new_state = self._apply_traffic(state=new_state)
+        if new_state != self._state:
+            self._state = new_state
+            self.present()
+
+    def _compute_nm_state(self) -> NetworkState:
+        """Return a NetworkState reflecting current NetworkManager status."""
         if not self._nm_client:
-            return
-
-        self._is_connected = False
-        self._is_wifi = False
-        self._ssid = ""
-        self._signal_strength = 0
-        self._iface = ""
-        self._ip_address = ""
+            return NetworkState()
 
         # Collect candidates, prioritize wifi > ethernet > other
         best_device: NM.Device | None = None
@@ -564,85 +558,109 @@ def _update_nm_state(self) -> None:
                     best_device = device
 
         if not best_device:
-            return
+            return NetworkState()
 
-        self._is_connected = True
-        self._iface = best_device.get_iface() or ""
-
-        # IP address
+        iface = best_device.get_iface() or ""
+        ip_address = ""
         ip4_config = best_device.get_ip4_config()
         if ip4_config:
             addrs = ip4_config.get_addresses()
             if addrs:
-                self._ip_address = addrs[0].get_address() or ""
+                ip_address = addrs[0].get_address() or ""
 
-        # WiFi specifics
+        is_wifi = False
+        ssid = ""
+        signal_strength = 0
         if isinstance(best_device, NM.DeviceWifi):
-            self._is_wifi = True
+            is_wifi = True
             ap = best_device.get_active_access_point()
             if ap:
                 ssid_bytes = ap.get_ssid()
                 if ssid_bytes:
-                    self._ssid = ssid_bytes.get_data().decode("utf-8", errors="replace")
-                self._signal_strength = ap.get_strength()
+                    ssid = ssid_bytes.get_data().decode("utf-8", errors="replace")
+                signal_strength = ap.get_strength()
+
+        return NetworkState(
+            is_connected=True,
+            is_wifi=is_wifi,
+            ssid=ssid,
+            signal_strength=signal_strength,
+            iface=iface,
+            ip_address=ip_address,
+            # Traffic speeds get filled in by _apply_traffic; keep previous values
+            # so equality comparison doesn't flap between NM refresh and tick.
+            rx_speed=self._state.rx_speed,
+            tx_speed=self._state.tx_speed,
+        )
 
     def _tick(self) -> bool:
-        """Poll traffic counters and wifi signal."""
-        self._update_nm_state()
-        self._update_traffic()
-        self._update_wifi_signal()
-        self.present()
+        """Poll traffic counters and wifi signal, re-render only on change."""
+        self._refresh_state()
         return True
 
-    def _update_traffic(self) -> None:
-        """Read /proc/net/dev and compute speeds for active interface."""
-        if not self._iface:
-            self._rx_speed = 0.0
-            self._tx_speed = 0.0
-            return
+    def _apply_traffic(self, *, state: NetworkState) -> NetworkState:
+        """Return ``state`` with rx/tx speeds updated from /proc/net/dev."""
+        if not state.iface:
+            self._prev_counters = None
+            return replace(state, rx_speed=0.0, tx_speed=0.0)
         try:
             with _PROC_NET_DEV.open() as f:
                 counters = parse_proc_net_dev(text=f.read())
         except OSError as exc:
             log.debug("Failed to read %s: %s", _PROC_NET_DEV, exc)
-            return
+            return state
 
         now = time.monotonic()
-        current = counters.get(self._iface)
+        current = counters.get(state.iface)
+        rx_speed = state.rx_speed
+        tx_speed = state.tx_speed
         if current and self._prev_counters:
             elapsed = now - self._prev_time
-            self._rx_speed, self._tx_speed = compute_speeds(
+            rx_speed, tx_speed = compute_speeds(
                 prev=self._prev_counters,
                 curr=current,
                 elapsed_s=elapsed,
             )
         if current:
             self._prev_counters = current
         self._prev_time = now
+        # NM may have stale signal strength; refresh while we have the client.
+        signal = self._refresh_wifi_signal(state=state)
+        return NetworkState(
+            is_connected=state.is_connected,
+            is_wifi=state.is_wifi,
+            ssid=state.ssid,
+            signal_strength=signal,
+            iface=state.iface,
+            ip_address=state.ip_address,
+            rx_speed=rx_speed,
+            tx_speed=tx_speed,
+        )
 
-    def _update_wifi_signal(self) -> None:
-        """Re-read wifi signal from NM (access point strength can change)."""
-        if not self._nm_client or not self._is_wifi:
-            return
+    def _refresh_wifi_signal(self, *, state: NetworkState) -> int:
+        """Return the current wifi signal strength, or ``state.signal_strength``."""
+        if not self._nm_client or not state.is_wifi:
+            return state.signal_strength
         for conn in self._nm_client.get_active_connections():
             if conn.get_state() != NM.ActiveConnectionState.ACTIVATED:
                 continue
             for device in conn.get_devices() or ():
                 if isinstance(device, NM.DeviceWifi):
                     ap = device.get_active_access_point()
                     if ap:
-                        self._signal_strength = ap.get_strength()
-                    return
+                        return ap.get_strength()
+                    return state.signal_strength
+        return state.signal_strength
 
     def _build_tooltip(self) -> str:
         return build_tooltip(
-            is_connected=self._is_connected,
-            ssid=self._ssid,
-            signal_strength=self._signal_strength,
-            iface=self._iface,
-            ip_address=self._ip_address,
-            rx_speed=self._rx_speed,
-            tx_speed=self._tx_speed,
+            is_connected=self._state.is_connected,
+            ssid=self._state.ssid,
+            signal_strength=self._state.signal_strength,
+            iface=self._state.iface,
+            ip_address=self._state.ip_address,
+            rx_speed=self._state.rx_speed,
+            tx_speed=self._state.tx_speed,
         )
 
     def _has_wifi_device(self) -> bool:

docking/applets/network/state.py

@@ -17,6 +17,7 @@
 
 import shutil
 import subprocess
+from dataclasses import dataclass
 from typing import NamedTuple
 
 from docking.applets.tooltip import structured_tooltip
@@ -51,6 +52,24 @@ class AvailableNetwork(NamedTuple):
     is_active: bool
 
 
+@dataclass(frozen=True)
+class NetworkState:
+    """Visible state of the Network applet.
+
+    Frozen so equality comparison gates re-rendering: the applet only calls
+    ``present()`` when this value actually changes between polls.
+    """
+
+    is_connected: bool = False
+    is_wifi: bool = False
+    ssid: str = ""
+    signal_strength: int = 0
+    iface: str = ""
+    ip_address: str = ""
+    rx_speed: float = 0.0
+    tx_speed: float = 0.0
+
+
 _CONNECTION_INFO_COMMANDS: tuple[tuple[str, ...], ...] = (
     ("gnome-control-center", "wifi"),
     ("gnome-control-center", "network"),