84 lines
2.2 KiB
Python
84 lines
2.2 KiB
Python
"""PiCopy – Blueprint: /api/wifi/*."""
|
||
|
||
import threading
|
||
import time
|
||
|
||
from flask import Blueprint, jsonify, request
|
||
|
||
from picopy.config import load_cfg, save_cfg
|
||
from picopy.wifi import (
|
||
wifi_state, wifi_lock,
|
||
is_ap_active, stop_ap, start_ap,
|
||
connect_client_wifi, update_wifi_state,
|
||
scan_wifi_networks,
|
||
)
|
||
|
||
wifi_bp = Blueprint('wifi', __name__)
|
||
|
||
|
||
@wifi_bp.route('/api/wifi/scan')
|
||
def r_wifi_scan():
|
||
nets = scan_wifi_networks()
|
||
return jsonify(nets)
|
||
|
||
|
||
@wifi_bp.route('/api/wifi/connect', methods=['POST'])
|
||
def r_wifi_connect():
|
||
data = request.get_json(force=True)
|
||
ssid = data.get('ssid', '').strip()
|
||
pw = data.get('password', '').strip()
|
||
if not ssid:
|
||
return jsonify(error='SSID fehlt'), 400
|
||
cfg = load_cfg()
|
||
cfg['wifi_ssid'] = ssid
|
||
cfg['wifi_password'] = pw
|
||
save_cfg(cfg)
|
||
|
||
def _connect():
|
||
ap_was_active = is_ap_active()
|
||
if ap_was_active:
|
||
stop_ap()
|
||
time.sleep(2)
|
||
ok = connect_client_wifi(ssid, pw)
|
||
if ok:
|
||
time.sleep(5)
|
||
update_wifi_state()
|
||
else:
|
||
if ap_was_active:
|
||
start_ap(cfg.get('ap_ssid', 'PiCopy'), cfg.get('ap_password', 'PiCopy,'))
|
||
update_wifi_state()
|
||
|
||
threading.Thread(target=_connect, daemon=True).start()
|
||
return jsonify(ok=True, msg='Verbindungsversuch gestartet')
|
||
|
||
|
||
@wifi_bp.route('/api/wifi/ap', methods=['POST'])
|
||
def r_wifi_ap():
|
||
data = request.get_json(force=True)
|
||
ssid = data.get('ssid', '').strip()
|
||
pw = data.get('password', '').strip()
|
||
if not ssid or len(pw) < 8:
|
||
return jsonify(error='SSID fehlt oder Passwort zu kurz (min. 8 Zeichen)'), 400
|
||
cfg = load_cfg()
|
||
cfg['ap_ssid'] = ssid
|
||
cfg['ap_password'] = pw
|
||
save_cfg(cfg)
|
||
|
||
def _restart_ap():
|
||
if is_ap_active():
|
||
stop_ap()
|
||
time.sleep(2)
|
||
start_ap(ssid, pw)
|
||
time.sleep(3)
|
||
with wifi_lock:
|
||
wifi_state.update(mode='ap', ssid=ssid, ip='10.42.0.1')
|
||
|
||
threading.Thread(target=_restart_ap, daemon=True).start()
|
||
return jsonify(ok=True)
|
||
|
||
|
||
@wifi_bp.route('/api/wifi/status')
|
||
def r_wifi_status():
|
||
with wifi_lock:
|
||
return jsonify(dict(wifi_state))
|