|
| 1 | +""" |
| 2 | +WiFi Service for MicroPythonOS. |
| 3 | +
|
| 4 | +Manages WiFi connections including: |
| 5 | +- Auto-connect to saved networks on boot |
| 6 | +- Network scanning |
| 7 | +- Connection management with saved credentials |
| 8 | +- Concurrent access locking |
| 9 | +
|
| 10 | +This service works alongside ConnectivityManager which monitors connection status. |
| 11 | +""" |
| 12 | + |
| 13 | +import ujson |
| 14 | +import os |
| 15 | +import time |
| 16 | + |
| 17 | +import mpos.config |
| 18 | +import mpos.time |
| 19 | + |
| 20 | +# Try to import network module (not available on desktop) |
| 21 | +HAS_NETWORK_MODULE = False |
| 22 | +try: |
| 23 | + import network |
| 24 | + HAS_NETWORK_MODULE = True |
| 25 | +except ImportError: |
| 26 | + print("WifiService: network module not available (desktop mode)") |
| 27 | + |
| 28 | + |
| 29 | +class WifiService: |
| 30 | + """ |
| 31 | + Service for managing WiFi connections. |
| 32 | +
|
| 33 | + This class handles connecting to saved WiFi networks and managing |
| 34 | + the WiFi hardware state. It's typically started in a background thread |
| 35 | + on boot to auto-connect to known networks. |
| 36 | + """ |
| 37 | + |
| 38 | + # Class-level lock to prevent concurrent WiFi operations |
| 39 | + # Used by WiFi app when scanning to avoid conflicts with connection attempts |
| 40 | + wifi_busy = False |
| 41 | + |
| 42 | + # Dictionary of saved access points {ssid: {password: "..."}} |
| 43 | + access_points = {} |
| 44 | + |
| 45 | + @staticmethod |
| 46 | + def connect(network_module=None): |
| 47 | + """ |
| 48 | + Scan for available networks and connect to the first saved network found. |
| 49 | +
|
| 50 | + Args: |
| 51 | + network_module: Network module for dependency injection (testing) |
| 52 | +
|
| 53 | + Returns: |
| 54 | + bool: True if successfully connected, False otherwise |
| 55 | + """ |
| 56 | + net = network_module if network_module else network |
| 57 | + wlan = net.WLAN(net.STA_IF) |
| 58 | + |
| 59 | + # Restart WiFi hardware in case it's in a bad state |
| 60 | + wlan.active(False) |
| 61 | + wlan.active(True) |
| 62 | + |
| 63 | + # Scan for available networks |
| 64 | + networks = wlan.scan() |
| 65 | + |
| 66 | + for n in networks: |
| 67 | + ssid = n[0].decode() |
| 68 | + print(f"WifiService: Found network '{ssid}'") |
| 69 | + |
| 70 | + if ssid in WifiService.access_points: |
| 71 | + password = WifiService.access_points.get(ssid).get("password") |
| 72 | + print(f"WifiService: Attempting to connect to saved network '{ssid}'") |
| 73 | + |
| 74 | + if WifiService.attempt_connecting(ssid, password, network_module=network_module): |
| 75 | + print(f"WifiService: Connected to '{ssid}'") |
| 76 | + return True |
| 77 | + else: |
| 78 | + print(f"WifiService: Failed to connect to '{ssid}'") |
| 79 | + else: |
| 80 | + print(f"WifiService: Skipping '{ssid}' (not configured)") |
| 81 | + |
| 82 | + print("WifiService: No saved networks found or connected") |
| 83 | + return False |
| 84 | + |
| 85 | + @staticmethod |
| 86 | + def attempt_connecting(ssid, password, network_module=None, time_module=None): |
| 87 | + """ |
| 88 | + Attempt to connect to a specific WiFi network. |
| 89 | +
|
| 90 | + Args: |
| 91 | + ssid: Network SSID to connect to |
| 92 | + password: Network password |
| 93 | + network_module: Network module for dependency injection (testing) |
| 94 | + time_module: Time module for dependency injection (testing) |
| 95 | +
|
| 96 | + Returns: |
| 97 | + bool: True if successfully connected, False otherwise |
| 98 | + """ |
| 99 | + print(f"WifiService: Connecting to SSID: {ssid}") |
| 100 | + |
| 101 | + net = network_module if network_module else network |
| 102 | + time_mod = time_module if time_module else time |
| 103 | + |
| 104 | + try: |
| 105 | + wlan = net.WLAN(net.STA_IF) |
| 106 | + wlan.connect(ssid, password) |
| 107 | + |
| 108 | + # Wait up to 10 seconds for connection |
| 109 | + for i in range(10): |
| 110 | + if wlan.isconnected(): |
| 111 | + print(f"WifiService: Connected to '{ssid}' after {i+1} seconds") |
| 112 | + |
| 113 | + # Sync time from NTP server if possible |
| 114 | + try: |
| 115 | + mpos.time.sync_time() |
| 116 | + except Exception as e: |
| 117 | + print(f"WifiService: Could not sync time: {e}") |
| 118 | + |
| 119 | + return True |
| 120 | + |
| 121 | + elif not wlan.active(): |
| 122 | + # WiFi was disabled during connection attempt |
| 123 | + print("WifiService: WiFi disabled during connection, aborting") |
| 124 | + return False |
| 125 | + |
| 126 | + print(f"WifiService: Waiting for connection, attempt {i+1}/10") |
| 127 | + time_mod.sleep(1) |
| 128 | + |
| 129 | + print(f"WifiService: Connection timeout for '{ssid}'") |
| 130 | + return False |
| 131 | + |
| 132 | + except Exception as e: |
| 133 | + print(f"WifiService: Connection error: {e}") |
| 134 | + return False |
| 135 | + |
| 136 | + @staticmethod |
| 137 | + def auto_connect(network_module=None, time_module=None): |
| 138 | + """ |
| 139 | + Auto-connect to a saved WiFi network on boot. |
| 140 | +
|
| 141 | + This is typically called in a background thread from main.py. |
| 142 | + It loads saved networks and attempts to connect to the first one found. |
| 143 | +
|
| 144 | + Args: |
| 145 | + network_module: Network module for dependency injection (testing) |
| 146 | + time_module: Time module for dependency injection (testing) |
| 147 | + """ |
| 148 | + print("WifiService: Auto-connect thread starting") |
| 149 | + |
| 150 | + # Load saved access points from config |
| 151 | + WifiService.access_points = mpos.config.SharedPreferences( |
| 152 | + "com.micropythonos.system.wifiservice" |
| 153 | + ).get_dict("access_points") |
| 154 | + |
| 155 | + if not len(WifiService.access_points): |
| 156 | + print("WifiService: No access points configured, exiting") |
| 157 | + return |
| 158 | + |
| 159 | + # Check if WiFi is busy (e.g., WiFi app is scanning) |
| 160 | + if WifiService.wifi_busy: |
| 161 | + print("WifiService: WiFi busy, auto-connect aborted") |
| 162 | + return |
| 163 | + |
| 164 | + WifiService.wifi_busy = True |
| 165 | + |
| 166 | + try: |
| 167 | + if not HAS_NETWORK_MODULE and network_module is None: |
| 168 | + # Desktop mode - simulate connection delay |
| 169 | + print("WifiService: Desktop mode, simulating connection...") |
| 170 | + time_mod = time_module if time_module else time |
| 171 | + time_mod.sleep(2) |
| 172 | + print("WifiService: Simulated connection complete") |
| 173 | + else: |
| 174 | + # Attempt to connect to saved networks |
| 175 | + if WifiService.connect(network_module=network_module): |
| 176 | + print("WifiService: Auto-connect successful") |
| 177 | + else: |
| 178 | + print("WifiService: Auto-connect failed") |
| 179 | + |
| 180 | + # Disable WiFi to conserve power if connection failed |
| 181 | + if network_module: |
| 182 | + net = network_module |
| 183 | + else: |
| 184 | + net = network |
| 185 | + |
| 186 | + wlan = net.WLAN(net.STA_IF) |
| 187 | + wlan.active(False) |
| 188 | + print("WifiService: WiFi disabled to conserve power") |
| 189 | + |
| 190 | + finally: |
| 191 | + WifiService.wifi_busy = False |
| 192 | + print("WifiService: Auto-connect thread finished") |
| 193 | + |
| 194 | + @staticmethod |
| 195 | + def is_connected(network_module=None): |
| 196 | + """ |
| 197 | + Check if WiFi is currently connected. |
| 198 | +
|
| 199 | + This is a simple connection check. For comprehensive connectivity |
| 200 | + monitoring with callbacks, use ConnectivityManager instead. |
| 201 | +
|
| 202 | + Args: |
| 203 | + network_module: Network module for dependency injection (testing) |
| 204 | +
|
| 205 | + Returns: |
| 206 | + bool: True if connected, False otherwise |
| 207 | + """ |
| 208 | + # If WiFi operations are in progress, report not connected |
| 209 | + if WifiService.wifi_busy: |
| 210 | + return False |
| 211 | + |
| 212 | + # Desktop mode - always report connected |
| 213 | + if not HAS_NETWORK_MODULE and network_module is None: |
| 214 | + return True |
| 215 | + |
| 216 | + # Check actual connection status |
| 217 | + try: |
| 218 | + net = network_module if network_module else network |
| 219 | + wlan = net.WLAN(net.STA_IF) |
| 220 | + return wlan.isconnected() |
| 221 | + except Exception as e: |
| 222 | + print(f"WifiService: Error checking connection: {e}") |
| 223 | + return False |
| 224 | + |
| 225 | + @staticmethod |
| 226 | + def disconnect(network_module=None): |
| 227 | + """ |
| 228 | + Disconnect from current WiFi network and disable WiFi. |
| 229 | +
|
| 230 | + Args: |
| 231 | + network_module: Network module for dependency injection (testing) |
| 232 | + """ |
| 233 | + if not HAS_NETWORK_MODULE and network_module is None: |
| 234 | + print("WifiService: Desktop mode, cannot disconnect") |
| 235 | + return |
| 236 | + |
| 237 | + try: |
| 238 | + net = network_module if network_module else network |
| 239 | + wlan = net.WLAN(net.STA_IF) |
| 240 | + wlan.disconnect() |
| 241 | + wlan.active(False) |
| 242 | + print("WifiService: Disconnected and WiFi disabled") |
| 243 | + except Exception as e: |
| 244 | + print(f"WifiService: Error disconnecting: {e}") |
| 245 | + |
| 246 | + @staticmethod |
| 247 | + def get_saved_networks(): |
| 248 | + """ |
| 249 | + Get list of saved network SSIDs. |
| 250 | +
|
| 251 | + Returns: |
| 252 | + list: List of saved SSIDs |
| 253 | + """ |
| 254 | + if not WifiService.access_points: |
| 255 | + WifiService.access_points = mpos.config.SharedPreferences( |
| 256 | + "com.micropythonos.system.wifiservice" |
| 257 | + ).get_dict("access_points") |
| 258 | + |
| 259 | + return list(WifiService.access_points.keys()) |
| 260 | + |
| 261 | + @staticmethod |
| 262 | + def save_network(ssid, password): |
| 263 | + """ |
| 264 | + Save a new WiFi network credential. |
| 265 | +
|
| 266 | + Args: |
| 267 | + ssid: Network SSID |
| 268 | + password: Network password |
| 269 | + """ |
| 270 | + # Load current saved networks |
| 271 | + prefs = mpos.config.SharedPreferences("com.micropythonos.system.wifiservice") |
| 272 | + access_points = prefs.get_dict("access_points") |
| 273 | + |
| 274 | + # Add or update the network |
| 275 | + access_points[ssid] = {"password": password} |
| 276 | + |
| 277 | + # Save back to config |
| 278 | + editor = prefs.edit() |
| 279 | + editor.put_dict("access_points", access_points) |
| 280 | + editor.commit() |
| 281 | + |
| 282 | + # Update class-level cache |
| 283 | + WifiService.access_points = access_points |
| 284 | + |
| 285 | + print(f"WifiService: Saved network '{ssid}'") |
| 286 | + |
| 287 | + @staticmethod |
| 288 | + def forget_network(ssid): |
| 289 | + """ |
| 290 | + Remove a saved WiFi network. |
| 291 | +
|
| 292 | + Args: |
| 293 | + ssid: Network SSID to forget |
| 294 | +
|
| 295 | + Returns: |
| 296 | + bool: True if network was found and removed, False otherwise |
| 297 | + """ |
| 298 | + # Load current saved networks |
| 299 | + prefs = mpos.config.SharedPreferences("com.micropythonos.system.wifiservice") |
| 300 | + access_points = prefs.get_dict("access_points") |
| 301 | + |
| 302 | + # Remove the network if it exists |
| 303 | + if ssid in access_points: |
| 304 | + del access_points[ssid] |
| 305 | + |
| 306 | + # Save back to config |
| 307 | + editor = prefs.edit() |
| 308 | + editor.put_dict("access_points", access_points) |
| 309 | + editor.commit() |
| 310 | + |
| 311 | + # Update class-level cache |
| 312 | + WifiService.access_points = access_points |
| 313 | + |
| 314 | + print(f"WifiService: Forgot network '{ssid}'") |
| 315 | + return True |
| 316 | + else: |
| 317 | + print(f"WifiService: Network '{ssid}' not found in saved networks") |
| 318 | + return False |
0 commit comments