"""Wi-Fi scanning, connection and status routes. Routes decorate the shared `api_v3` Blueprint from the package `__init__`, so their endpoint names are unchanged by living here. """ import threading import time from web_interface.blueprints.api_v3 import ( api_v3, describe_exception, jsonify, logger, request, ) # Joining a network while the setup AP is up has to take the AP down first, # and the caller is almost always a phone connected *through* that AP. Run # synchronously, the request outlives the link it arrived on: the browser # never gets an answer and the Connect button appears to do nothing. So in AP # mode the connect runs in the background and the result is kept here, where # /wifi/status can report it once the user is back on either network. _connect_lock = threading.Lock() _last_connect_attempt = None # Lets the 202 reach the phone before hostapd stops. _AP_HANDOFF_DELAY_SECONDS = 2.0 def _last_connect_snapshot(): with _connect_lock: return dict(_last_connect_attempt) if _last_connect_attempt else None def _spawn(target): threading.Thread(target=target, name='wifi-connect', daemon=True).start() def _connect_result_payload(ssid, success, message): """Shape a connect_to_network result for the client. The manager flags a rejected passphrase by prefixing the message with "wrong_password:"; the setup pages key off `error_type` rather than parsing nmcli's wording. """ if success: return {'status': 'success', 'message': message} payload = {'status': 'error', 'message': message or 'Failed to connect to network'} if message and message.startswith('wrong_password:'): payload['error_type'] = 'wrong_password' payload['message'] = f'Incorrect password for {ssid}' return payload def _record_connect_result(ssid, payload): global _last_connect_attempt with _connect_lock: _last_connect_attempt = { 'ssid': ssid, 'state': 'success' if payload['status'] == 'success' else 'failed', 'message': payload['message'], 'error_type': payload.get('error_type'), 'finished_at': time.time(), } def _run_background_connect(ssid, password): try: time.sleep(_AP_HANDOFF_DELAY_SECONDS) from src.wifi_manager import WiFiManager success, message = WiFiManager().connect_to_network(ssid, password) payload = _connect_result_payload(ssid, success, message) except Exception as e: logger.error("Background WiFi connect failed", exc_info=True) payload = {'status': 'error', 'message': describe_exception(e)} _record_connect_result(ssid, payload) def _parse_bool_ish(value): """Coerce a JSON value that is supposed to be a boolean. A JSON boolean arrives as a real Python bool, but these routes are a public HTTP contract and not every caller sends one. `bool(value)` gets two common cases wrong: `bool("false")` is True (a non-empty string is always truthy), and a plain int does not match an `is True` check (`1 is True` is False, since `True` is a distinct singleton from the int `1`) -- so a caller sending `{"enabled": 1}` was silently treated as False. Recognizes a real bool, "true"/"false"/"1"/"0"/"yes"/"no" case-insensitively, and int 1/0. Returns None for anything else, rather than guessing. A supplied-but- unrecognized value (e.g. a typo) used to silently become False here, which for `enabled` on the radio route could disconnect Wi-Fi the caller never asked to turn off -- callers must treat None as a validation error, not a real False. """ if isinstance(value, bool): return value if isinstance(value, str): lowered = value.strip().lower() if lowered in ('true', '1', 'yes'): return True if lowered in ('false', '0', 'no'): return False return None if isinstance(value, int): if value == 1: return True if value == 0: return False return None return None # WiFi Management Endpoints @api_v3.route('/wifi/status', methods=['GET']) def get_wifi_status(): """Get current WiFi connection status""" from src.wifi_manager import WiFiManager wifi_manager = WiFiManager() status = wifi_manager.get_wifi_status() # Get auto-enable setting from config auto_enable_ap = wifi_manager.config.get("auto_enable_ap_mode", True) # Default: True (safe due to grace period) return jsonify({ 'status': 'success', 'data': { 'connected': status.connected, 'ssid': status.ssid, 'ip_address': status.ip_address, 'signal': status.signal, 'ap_mode_active': status.ap_mode_active, 'auto_enable_ap_mode': auto_enable_ap, 'last_connect_attempt': _last_connect_snapshot(), } }) @api_v3.route('/wifi/scan', methods=['GET']) def scan_wifi_networks(): """Scan for available WiFi networks If AP mode is active, it will be temporarily disabled during scanning and automatically re-enabled afterward. Users connected to the AP will be briefly disconnected during this process. """ try: from src.wifi_manager import WiFiManager wifi_manager = WiFiManager() # Check if AP mode is active before scanning (for user notification) ap_was_active = wifi_manager._is_ap_mode_active() # Perform the scan (this will handle AP mode disabling/enabling internally) networks, _was_cached = wifi_manager.scan_networks() # Convert to dict format networks_data = [ { 'ssid': net.ssid, 'signal': net.signal, 'security': net.security, 'frequency': net.frequency } for net in networks ] response_data = { 'status': 'success', 'data': networks_data } # Inform user if AP mode was temporarily disabled if ap_was_active: response_data['message'] = ( f'Found {len(networks_data)} networks. ' 'Note: AP mode was temporarily disabled during scanning and has been re-enabled. ' 'If you were connected to the setup network, you may need to reconnect.' ) return jsonify(response_data) except Exception as e: logger.error("Error scanning WiFi networks", exc_info=True) error_message = 'An error occurred while scanning WiFi networks; see logs for details' # Provide more specific error messages for common issues error_str = str(e).lower() if 'permission' in error_str or 'sudo' in error_str: error_message = ( 'Permission error while scanning. ' 'The WiFi scan requires appropriate permissions. ' 'Please ensure the application has necessary privileges.' ) elif 'timeout' in error_str: error_message = ( 'WiFi scan timed out. ' 'The scan took too long to complete. ' 'This may happen if the WiFi interface is busy or in use.' ) elif 'no wifi' in error_str or 'not available' in error_str: error_message = ( 'WiFi scanning tools are not available. ' 'Please ensure NetworkManager (nmcli) or iwlist is installed.' ) return jsonify({ 'status': 'error', 'message': error_message }), 500 @api_v3.route('/wifi/connect', methods=['POST']) def connect_wifi(): """Connect to a WiFi network. With the setup AP active this answers 202 at once and connects in the background (see _last_connect_attempt); otherwise it waits for the result. """ global _last_connect_attempt from src.wifi_manager import WiFiManager data = request.get_json(silent=True) if not data: return jsonify({ 'status': 'error', 'message': 'Request body is required' }), 400 if 'ssid' not in data: return jsonify({ 'status': 'error', 'message': 'SSID is required' }), 400 ssid = data['ssid'] if not ssid or not ssid.strip(): return jsonify({ 'status': 'error', 'message': 'SSID cannot be empty' }), 400 ssid = ssid.strip() password = data.get('password', '') or '' wifi_manager = WiFiManager() ap_mode_active = wifi_manager._is_ap_mode_active() # One attempt at a time on either path: concurrent connects fight over # the radio, and the first to finish would clear the in-progress flag # the monitor daemon still needs for the other. The check can't depend # on AP state either -- a background attempt takes the AP down long # before it finishes. with _connect_lock: if _last_connect_attempt and _last_connect_attempt['state'] == 'pending': return jsonify({ 'status': 'error', 'message': f"Already connecting to {_last_connect_attempt['ssid']}" }), 409 _last_connect_attempt = { 'ssid': ssid, 'state': 'pending', 'message': None, 'error_type': None, 'finished_at': None, } if ap_mode_active: try: _spawn(lambda: _run_background_connect(ssid, password)) except Exception: # Nothing will ever finish this attempt; don't leave every # later request refused. with _connect_lock: _last_connect_attempt = None raise return jsonify({ 'status': 'pending', 'message': ( f'Connecting to {ssid}. The LEDMatrix-Setup network will turn off, ' 'so this page will lose its connection.' ), 'data': {'ssid': ssid}, }), 202 try: success, message = wifi_manager.connect_to_network(ssid, password) except Exception as e: _record_connect_result(ssid, {'status': 'error', 'message': describe_exception(e)}) raise payload = _connect_result_payload(ssid, success, message) _record_connect_result(ssid, payload) return jsonify(payload), (200 if success else 400) @api_v3.route('/wifi/disconnect', methods=['POST']) def disconnect_wifi(): """Disconnect from the current WiFi network""" from src.wifi_manager import WiFiManager wifi_manager = WiFiManager() success, message = wifi_manager.disconnect_from_network() if success: return jsonify({ 'status': 'success', 'message': message }) else: return jsonify({ 'status': 'error', 'message': message or 'Failed to disconnect from network' }), 400 @api_v3.route('/wifi/ap/enable', methods=['POST']) def enable_ap_mode(): """Enable access point mode""" from src.wifi_manager import WiFiManager wifi_manager = WiFiManager() # The same parsing as every other boolean on these routes. Anything it # does not recognise (None) is not a request to force. force = _parse_bool_ish((request.get_json(silent=True) or {}).get('force', False)) is True success, message = wifi_manager.enable_ap_mode(force=force) if success: return jsonify({ 'status': 'success', 'message': message }) else: return jsonify({ 'status': 'error', 'message': message }), 400 @api_v3.route('/wifi/ap/disable', methods=['POST']) def disable_ap_mode(): """Disable access point mode""" from src.wifi_manager import WiFiManager wifi_manager = WiFiManager() success, message = wifi_manager.disable_ap_mode() if success: return jsonify({ 'status': 'success', 'message': message }) else: return jsonify({ 'status': 'error', 'message': message }), 400 @api_v3.route('/wifi/ap/auto-enable', methods=['GET']) def get_auto_enable_ap_mode(): """Get auto-enable AP mode setting""" from src.wifi_manager import WiFiManager wifi_manager = WiFiManager() auto_enable = wifi_manager.config.get("auto_enable_ap_mode", True) # Default: True (safe due to grace period) return jsonify({ 'status': 'success', 'data': { 'auto_enable_ap_mode': auto_enable } }) @api_v3.route('/wifi/ap/auto-enable', methods=['POST']) def set_auto_enable_ap_mode(): """Set auto-enable AP mode setting""" from src.wifi_manager import WiFiManager data = request.get_json(silent=True) if data is None or 'auto_enable_ap_mode' not in data: return jsonify({ 'status': 'error', 'message': 'auto_enable_ap_mode is required' }), 400 auto_enable = _parse_bool_ish(data['auto_enable_ap_mode']) if auto_enable is None: return jsonify({ 'status': 'error', 'message': 'auto_enable_ap_mode must be a boolean' }), 400 wifi_manager = WiFiManager() wifi_manager.config["auto_enable_ap_mode"] = auto_enable if not wifi_manager._save_config(): return jsonify({ 'status': 'error', 'message': (f'Could not save the setting to {wifi_manager.config_path}; ' 'check that the web interface user can write it.'), }), 500 return jsonify({ 'status': 'success', 'message': f'Auto-enable AP mode set to {auto_enable}', 'data': { 'auto_enable_ap_mode': auto_enable } }) @api_v3.route('/wifi/radio', methods=['GET']) def get_wifi_radio(): """Get current WiFi radio state (enabled/disabled) and wired-fallback status.""" from src.wifi_manager import WiFiManager wifi_manager = WiFiManager() state = wifi_manager.get_wifi_radio_state() return jsonify({ 'status': 'success', 'data': state }) @api_v3.route('/wifi/radio', methods=['POST']) def set_wifi_radio(): """Turn the WiFi radio on or off. Body: {"enabled": bool, "force": bool (optional)}. Disabling is refused unless Ethernet is connected or force=True, to avoid locking the user out of this web interface. """ from src.wifi_manager import WiFiManager data = request.get_json(silent=True) or {} if 'enabled' not in data: return jsonify({ 'status': 'error', 'message': 'enabled is required' }), 400 # Parse defensively: bool("false") is True and a plain int never # matches `is True`, so `_parse_bool_ish` handles bool, string and # int 1/0 — the endpoint is a public contract, not just the shipped # UI (which always sends real JSON booleans). An unrecognized value # must be rejected, not silently disable the radio: this is the # route that can drop the caller's own connection to this interface. enabled = _parse_bool_ish(data['enabled']) if enabled is None: return jsonify({ 'status': 'error', 'message': 'enabled must be a boolean' }), 400 force = _parse_bool_ish(data.get('force', False)) if force is None: return jsonify({ 'status': 'error', 'message': 'force must be a boolean' }), 400 wifi_manager = WiFiManager() success, message, reason = wifi_manager.set_wifi_radio(enabled, force=force) if success: return jsonify({ 'status': 'success', 'message': message, 'data': wifi_manager.get_wifi_radio_state() }) else: return jsonify({ 'status': 'error', 'message': message, 'reason': reason }), 400