From b4237f2f1df797cef365bda6f9fb350bc9ddd08b Mon Sep 17 00:00:00 2001 From: Shawn Date: Tue, 9 Jun 2026 01:14:41 -0400 Subject: [PATCH] Add Network + Bluetooth config in Web UI - src/system/network.py: WiFi scan/connect/hotspot via nmcli + iwlist fallback - src/system/bluetooth.py: BT scan/pair/connect/MIDI via bluetoothctl - REST endpoints: /api/wifi/* and /api/bluetooth/* (16 endpoints) - Web UI: tabbed settings with Network + Bluetooth panels - network.js: WiFi scan, connect modal, saved nets, hotspot - bluetooth.js: BT scan, pair/unpair, connect, MIDI service - style.css: tabs, network list, BT device list, signal indicators - scripts/setup-bt-midi.sh: BT MIDI systemd bridge service - tests/test_network.py: 22 tests (nmcli/iwlist, fallback, hotspot) - tests/test_bt.py: 24 tests (status, scan, pair, MIDI, edge cases) - _gather_state includes wifi + bluetooth sub-objects - All 93 tests pass --- scripts/setup-bt-midi.sh | 285 +++++++++------ src/system/bluetooth.py | 438 ++++++++++++++++++++++ src/system/network.py | 620 ++++++++++++++++++++++++++++++++ src/web/server.py | 214 +++++++++++ src/web/static/bluetooth.js | 294 +++++++++++++++ src/web/static/network.js | 259 +++++++++++++ src/web/static/style.css | 300 ++++++++++++++++ src/web/templates/settings.html | 375 ++++++++++++++----- tests/test_bt.py | 383 ++++++++++++++++++++ tests/test_network.py | 323 +++++++++++++++++ 10 files changed, 3305 insertions(+), 186 deletions(-) create mode 100644 src/system/bluetooth.py create mode 100644 src/system/network.py create mode 100644 src/web/static/bluetooth.js create mode 100644 src/web/static/network.js create mode 100644 tests/test_bt.py create mode 100644 tests/test_network.py diff --git a/scripts/setup-bt-midi.sh b/scripts/setup-bt-midi.sh index c70b1bd..cc6656e 100644 --- a/scripts/setup-bt-midi.sh +++ b/scripts/setup-bt-midi.sh @@ -1,16 +1,23 @@ #!/usr/bin/env bash -# ── Pi Multi-FX Pedal — Bluetooth MIDI Setup ─────────────────────── -# Enables Bluetooth MIDI (BLE-MIDI) so wireless controllers, -# expression pedals, and phone apps can send MIDI to the pedal. +# ── Pi Multi-FX Pedal — Bluetooth MIDI Service Setup ────────────── +# Configures BlueZ for Bluetooth MIDI (BLE MIDI) on the Pi. +# +# This sets up: +# 1. BlueZ ALSA MIDI bridge (bluez-alsa-utils for classic BT MIDI) +# 2. A systemd service that auto-starts the BT MIDI bridge on boot # # Usage: -# sudo bash scripts/setup-bt-midi.sh # Interactive setup -# sudo bash scripts/setup-bt-midi.sh --disable # Remove BT MIDI service -# sudo bash scripts/setup-bt-midi.sh --pair-only # Just make it discoverable +# sudo bash scripts/setup-bt-midi.sh # Install + enable +# sudo bash scripts/setup-bt-midi.sh --uninstall # Remove service # # Requirements: -# Built-in RPi Bluetooth, or USB BT dongle -# bluez, bluez-utils (installed by this script) +# - Raspberry Pi with built-in Bluetooth (Pi 3/4/5/Zero 2W) +# - bluez installed (usually present on RPi OS / DietPi) +# +# After setup: +# - Systemd service 'pi-bt-midi' is enabled and starts on boot +# - Bluetooth MIDI devices can pair via the web UI +# - MIDI data flows through ALSA sequencer (port 14:0) # ──────────────────────────────────────────────────────────────────── set -euo pipefail @@ -18,141 +25,207 @@ set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" PROJECT_DIR="$(dirname "$SCRIPT_DIR")" -DISABLE=false -PAIR_ONLY=false +# ── Colors ──────────────────────────────────────────────────────────── +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +CYAN='\033[0;36m' +NC='\033[0m' -while [[ $# -gt 0 ]]; do - case "$1" in - --disable) DISABLE=true; shift ;; - --pair-only) PAIR_ONLY=true; shift ;; - --help|-h) echo "Usage: $0 [--disable] [--pair-only]"; exit 0 ;; - *) echo "Unknown: $1"; exit 1 ;; - esac -done - -RED='\033[0;31m'; GREEN='\033[0;32m'; YELLOW='\033[1;33m'; CYAN='\033[0;36m'; NC='\033[0m' info() { echo -e "${CYAN}[INFO]${NC} $1"; } ok() { echo -e "${GREEN}[OK]${NC} $1"; } warn() { echo -e "${YELLOW}[WARN]${NC} $1"; } fail() { echo -e "${RED}[FAIL]${NC} $1"; } -if [ "$DISABLE" = true ]; then - echo "" - info "Disabling Bluetooth MIDI..." +# ── Config ──────────────────────────────────────────────────────────── +SERVICE_NAME="pi-bt-midi" +SERVICE_PATH="/etc/systemd/system/${SERVICE_NAME}.service" - sudo systemctl stop bluetooth-midi 2>/dev/null || true - sudo systemctl disable bluetooth-midi 2>/dev/null || true - sudo rm -f /etc/systemd/system/bluetooth-midi.service +# ── Parse args ──────────────────────────────────────────────────────── +UNINSTALL=false +while [[ $# -gt 0 ]]; do + case "$1" in + --uninstall) UNINSTALL=true; shift ;; + --help|-h) echo "Usage: $0 [--uninstall]"; exit 0 ;; + *) echo "Unknown: $1"; exit 1 ;; + esac +done - # Restart bluetooth in standard mode - sudo systemctl restart bluetooth - ok "Bluetooth MIDI disabled" +# ═══════════════════════════════════════════════════════════════════════ +# UNINSTALL +# ═══════════════════════════════════════════════════════════════════════ + +if [ "$UNINSTALL" = true ]; then + info "Removing Bluetooth MIDI service..." + + sudo systemctl stop "${SERVICE_NAME}" 2>/dev/null || true + sudo systemctl disable "${SERVICE_NAME}" 2>/dev/null || true + sudo rm -f "$SERVICE_PATH" + sudo systemctl daemon-reload + + # Remove the bt-midi script if it was copied + sudo rm -f /usr/local/bin/bt-midi-bridge.sh 2>/dev/null || true + + ok "Bluetooth MIDI service removed." exit 0 fi +# ═══════════════════════════════════════════════════════════════════════ +# INSTALL +# ═══════════════════════════════════════════════════════════════════════ + echo "" echo "╔═══════════════════════════════════════════════╗" echo "║ Pi Multi-FX Pedal — Bluetooth MIDI Setup ║" echo "╚═══════════════════════════════════════════════╝" echo "" -# 1. Install bluez -info "Installing Bluetooth packages..." -sudo apt-get install -y -qq bluez bluez-utils bluez-tools 2>/dev/null || true -ok "Bluetooth packages installed" +# ── 1. Check prerequisites ──────────────────────────────────────────── -# 2. Ensure BT hardware is on -info "Enabling Bluetooth hardware..." -sudo raspi-gpio set 28 a0 2>/dev/null || true # Enable BT on RPi -sudo hciconfig hci0 up 2>/dev/null || true -ok "Bluetooth hardware enabled" +info "Checking prerequisites..." -# 3. Enable BT MIDI plugin in bluez -info "Configuring BlueZ for MIDI..." -BLUEZ_CONF="/etc/bluetooth/main.conf" -if [ -f "$BLUEZ_CONF" ]; then - sudo sed -i 's/^#Class *= *0x000100/Class = 0x002300/' "$BLUEZ_CONF" 2>/dev/null || true - sudo sed -i 's/^#DiscoverableTimeout *= *0/DiscoverableTimeout = 0/' "$BLUEZ_CONF" 2>/dev/null || true - sudo sed -i 's/^#PairableTimeout *= *0/PairableTimeout = 0/' "$BLUEZ_CONF" 2>/dev/null || true -fi -ok "BlueZ configured for discoverable + pairable MIDI" - -# 4. Make device discoverable -info "Making pedal discoverable as 'Pi-Pedal-MIDI'..." -cat </dev/null; then + fail "bluetoothctl not found. Install bluez: sudo apt install bluez" + exit 1 fi -# 5. Create systemd service for BT MIDI -info "Creating bluetooth-midi systemd service..." +if ! systemctl is-active --quiet bluetooth; then + warn "Bluetooth service is not running. Starting it..." + sudo systemctl start bluetooth + sleep 1 +fi + +ok "BlueZ is available" + +# ── 2. Install bluez-alsa-utils for ALSA MIDI bridge ───────────────── + +if ! command -v bluealsa &>/dev/null; then + info "Installing bluez-alsa-utils for ALSA MIDI bridge..." + sudo apt-get update -qq + sudo apt-get install -y -qq bluez-alsa-utils + ok "bluez-alsa-utils installed" +else + ok "bluez-alsa-utils already installed" +fi + +# ── 3. Create the MIDI bridge helper script ─────────────────────────── + +info "Creating bt-midi-bridge helper..." + +sudo tee /usr/local/bin/bt-midi-bridge.sh > /dev/null << 'BRIDGESCRIPT' +#!/usr/bin/env bash +# ── Bluetooth MIDI Bridge ────────────────────────────────────────── +# Launched by systemd (pi-bt-midi.service). +# Keeps bluealsa MIDI running and handles reconnection. +# +# The ALSA sequencer port for BlueALSA MIDI is typically 14:0. +# Our pedal app reads MIDI from this port. + +set -euo pipefail + +# Log to systemd journal +exec > >(logger -t bt-midi-bridge) 2>&1 + +echo "Starting Bluetooth MIDI bridge..." + +# Ensure Bluetooth is powered on +bluetoothctl power on + +# Start bluealsa service (MIDI profiles) +if command -v bluealsa &>/dev/null; then + # bluealsa runs as a system service; just verify it's running + if systemctl is-active --quiet bluealsa; then + echo "bluealsa is running" + else + echo "Starting bluealsa..." + sudo systemctl start bluealsa + fi +fi + +# Register with ALSA sequencer for MIDI connections +# This makes our app visible as an ALSA MIDI port +if command -v aconnect &>/dev/null; then + echo "ALSA sequencer available" + # List MIDI ports for debugging + aconnect -i -o +fi + +echo "Bluetooth MIDI bridge ready." + +# Keep running (systemd will restart if needed) +while true; do + # Check Bluetooth health and reconnect if needed + if ! bluetoothctl show | grep -q "Powered: yes"; then + echo "BT not powered, re-powering..." + bluetoothctl power on + fi + sleep 30 +done +BRIDGESCRIPT + +sudo chmod +x /usr/local/bin/bt-midi-bridge.sh +ok "Bridge script created at /usr/local/bin/bt-midi-bridge.sh" + +# ── 4. Create systemd service ───────────────────────────────────────── + +info "Creating systemd service..." + +sudo tee "$SERVICE_PATH" > /dev/null << SERVICECONTENT +# Pi Multi-FX Pedal — Bluetooth MIDI Bridge +# Installed by scripts/setup-bt-midi.sh +# Auto-starts BlueALSA MIDI bridge for wireless MIDI controllers -cat <<'SERVICE' | sudo tee /etc/systemd/system/bluetooth-midi.service > /dev/null [Unit] -Description=Pi Multi-FX Pedal — Bluetooth MIDI Service -After=bluetooth.service -Requires=bluetooth.service +Description=Pi Multi-FX Pedal — Bluetooth MIDI Bridge +Documentation=https://gitea.ourpad.casa/shawn/pi-multifx-pedal +After=bluetooth.service bluealsa.service +Wants=bluetooth.service +BindsTo=multi-fx-pedal.target [Service] -Type=oneshot -RemainAfterExit=yes -ExecStart=/usr/bin/bluetoothctl discoverable on -ExecStartPost=/usr/bin/bluetoothctl pairable on -ExecStartPost=/usr/bin/bluetoothctl name Pi-Pedal-MIDI -ExecStop=/usr/bin/bluetoothctl discoverable off -Restart=no +Type=simple +User=pi +Group=audio +ExecStart=/usr/local/bin/bt-midi-bridge.sh +Restart=on-failure +RestartSec=5 +TimeoutStopSec=10 [Install] -WantedBy=multi-user.target -SERVICE +WantedBy=multi-fx-pedal.target +multi-user.target +SERVICECONTENT -# 6. Register MIDI over BLE with ALSA (a2dp-midi bridge) -# On modern BlueZ (5.50+), BLE-MIDI shows up as a sequencer port automatically -# Verify with: aconnect -o -l | grep -i midi +ok "Service unit created at ${SERVICE_PATH}" +# ── 5. Enable and start ─────────────────────────────────────────────── + +info "Enabling and starting Bluetooth MIDI service..." sudo systemctl daemon-reload -sudo systemctl enable bluetooth-midi -sudo systemctl start bluetooth-midi +sudo systemctl enable "${SERVICE_NAME}" +sudo systemctl start "${SERVICE_NAME}" 2>/dev/null || true -# 7. Also start the pedal's MIDI handler to accept BT MIDI -# The MIDI handler already listens on ALSA sequencer ports -info "MIDI handler will auto-detect BT MIDI sequencer ports on next pedal start" +sleep 1 -ok "Bluetooth MIDI service started" +if systemctl is-active --quiet "${SERVICE_NAME}"; then + ok "Bluetooth MIDI service is running" +else + warn "Service created but not yet active. Check: sudo systemctl status ${SERVICE_NAME}" +fi + +# ═══════════════════════════════════════════════════════════════════════ +# DONE +# ═══════════════════════════════════════════════════════════════════════ echo "" echo "╔═══════════════════════════════════════════════╗" echo "║ Bluetooth MIDI Setup Complete 🎸 ║" echo "╚═══════════════════════════════════════════════╝" echo "" -echo " 📡 Name: Pi-Pedal-MIDI" -echo " 🎛 Connect from: Phone, tablet, wireless footswitch" +echo " 🎵 BT MIDI bridge: ${SERVICE_NAME}" +echo " 📋 Service status: sudo systemctl status ${SERVICE_NAME}" +echo " 🔗 Logs: sudo journalctl -u ${SERVICE_NAME} -n 30 --no-pager" echo "" -echo " From your phone:" -echo " 1. Enable Bluetooth" -echo " 2. Scan for 'Pi-Pedal-MIDI'" -echo " 3. Pair" -echo " 4. Open any MIDI controller app" -echo "" -echo " To disable later:" -echo " sudo bash $0 --disable" +echo "To remove:" +echo " sudo bash $0 --uninstall" echo "" \ No newline at end of file diff --git a/src/system/bluetooth.py b/src/system/bluetooth.py new file mode 100644 index 0000000..a531ab5 --- /dev/null +++ b/src/system/bluetooth.py @@ -0,0 +1,438 @@ +"""Bluetooth management module for Pi Multi-FX Pedal. + +Uses bluetoothctl (BlueZ) for all Bluetooth operations — no extra deps +on RPi OS or DietPi. Handles device scanning, pairing, unpairing, and +Bluetooth MIDI service management via systemd. +""" + +from __future__ import annotations + +import asyncio +import logging +import re +import subprocess +import time +from pathlib import Path +from typing import Any, Optional + +logger = logging.getLogger(__name__) + +# ── Constants ───────────────────────────────────────────────────────────────── + +BT_MIDI_SERVICE_NAME = "pi-bt-midi" +BT_MIDI_SERVICE_PATH = Path(f"/etc/systemd/system/{BT_MIDI_SERVICE_NAME}.service") +BTCTL_TIMEOUT = 30 # seconds to wait for bluetoothctl commands + +# ── Low-level bluetoothctl helpers ──────────────────────────────────────────── + + +def _bluetoothctl(cmd: str, timeout: int = BTCTL_TIMEOUT) -> str: + """Run a bluetoothctl command and return stdout. + + Uses the simple command form: bluetoothctl + """ + try: + result = subprocess.run( + ["bluetoothctl", cmd], + capture_output=True, text=True, timeout=timeout, + ) + return result.stdout + except FileNotFoundError: + logger.warning("bluetoothctl not found — BlueZ not installed") + return "" + except subprocess.TimeoutExpired: + logger.warning("bluetoothctl command timed out: %s", cmd) + return "" + except Exception as e: + logger.warning("bluetoothctl error: %s", e) + return "" + + +def _bluetoothctl_interactive(commands: list[str], timeout: int = 15) -> str: + """Run multiple commands in an interactive bluetoothctl session. + + Uses echo-pipe to feed commands into bluetoothctl. + """ + try: + full_cmd = " && ".join( + f"echo '{c}'" for c in commands[:-1] + ) + f" && echo '{commands[-1]}'" + result = subprocess.run( + ["bash", "-c", f"{full_cmd} | bluetoothctl"], + capture_output=True, text=True, timeout=timeout, + ) + return result.stdout + except Exception as e: + logger.warning("bluetoothctl interactive error: %s", e) + return "" + + +def _bt_available() -> bool: + """Check if Bluetooth hardware is available and not blocked.""" + try: + result = subprocess.run( + ["bluetoothctl", "show"], + capture_output=True, text=True, timeout=10, + ) + if result.returncode != 0: + return False + # Check if powered on + return "Powered: yes" in result.stdout + except FileNotFoundError: + return False + + +# ── Core Bluetooth operations ───────────────────────────────────────────────── + + +def bt_status() -> dict[str, Any]: + """Get Bluetooth adapter status. + + Returns dict with: available (bool), powered (bool), name (str), + discoverable (bool), pairable (bool), address (str|None). + """ + output = _bluetoothctl("show", timeout=10) + if not output: + return { + "available": False, "powered": False, + "name": None, "discoverable": False, "pairable": False, + "address": None, + } + + available = True + powered = "Powered: yes" in output + name = None + discoverable = "Discoverable: yes" in output + pairable = "Pairable: yes" in output + address = None + + for line in output.split("\n"): + line = line.strip() + if line.startswith("Name:"): + name = line.split(":", 1)[1].strip() + elif line.startswith("Controller"): + parts = line.split() + if len(parts) >= 2 and ":" in parts[1]: + address = parts[1] + + return { + "available": available, + "powered": powered, + "name": name or "pi-pedal", + "discoverable": discoverable, + "pairable": pairable, + "address": address, + } + + +def bt_power(on: bool = True) -> dict[str, Any]: + """Power Bluetooth adapter on or off. + + Args: + on: True to power on, False to power off. + + Returns: + dict with ok (bool) key. + """ + cmd = "power on" if on else "power off" + output = _bluetoothctl(cmd, timeout=15) + # Verify + time.sleep(0.5) + status = bt_status() + success = status.get("powered") == on + logger.info("Bluetooth power %s: %s", "on" if on else "off", "OK" if success else "FAIL") + return {"ok": success} + + +def bt_discoverable(on: bool = True) -> dict[str, Any]: + """Set Bluetooth discoverable mode. + + Args: + on: True to make discoverable, False to hide. + + Returns: + dict with ok (bool) key. + """ + cmd = "discoverable on" if on else "discoverable off" + _bluetoothctl(cmd, timeout=15) + time.sleep(0.5) + status = bt_status() + success = status.get("discoverable") == on + return {"ok": success} + + +def bt_scan(timeout: int = 12) -> dict[str, Any]: + """Scan for discoverable Bluetooth devices. + + Uses bluetoothctl scan on, waits, then parses the device list. + + Args: + timeout: Seconds to scan (default 12). + + Returns: + dict with ok (bool) and devices (list of dicts with address, name, rssi). + """ + # Start scanning + output = _bluetoothctl("scan on", timeout=timeout + 5) + + # Wait for devices to appear + time.sleep(timeout) + + # Stop scanning + _bluetoothctl("scan off", timeout=5) + + # Get the device list + devices_output = _bluetoothctl("devices", timeout=5) + devices: list[dict[str, Any]] = [] + + # Parse devices (format: Device AA:BB:CC:DD:EE:FF Device Name) + device_pattern = re.compile(r"Device\s+([0-9A-Fa-f:]+)\s+(.+)$") + for line in devices_output.split("\n"): + match = device_pattern.match(line.strip()) + if match: + addr = match.group(1).upper() + name = match.group(2).strip() + # Try to find RSSI info from the scan output + rssi = None + for scan_line in output.split("\n"): + if addr in scan_line and "RSSI" in scan_line: + rssi_match = re.search(r"RSSI[:\s]+(-?\d+)", scan_line) + if rssi_match: + rssi = int(rssi_match.group(1)) + break + devices.append({ + "address": addr, + "name": name if name else "Unknown", + "rssi": rssi, + }) + + # Sort by RSSI (strongest first) + devices.sort(key=lambda d: d["rssi"] if d["rssi"] is not None else -999, reverse=True) + + logger.info("BT scan found %d devices", len(devices)) + return {"ok": True, "devices": devices} + + +def bt_paired_devices() -> list[dict[str, Any]]: + """List currently paired Bluetooth devices. + + Returns list of dicts with address, name, connected, trusted. + """ + output = _bluetoothctl("paired-devices", timeout=10) + devices: list[dict[str, Any]] = [] + + # Get connected devices to mark them + info_output = _bluetoothctl("info", timeout=10) + + pattern = re.compile(r"Device\s+([0-9A-Fa-f:]+)\s+(.+)$") + for line in output.split("\n"): + match = pattern.match(line.strip()) + if match: + addr = match.group(1).upper() + name = match.group(2).strip() + # Check if connected + connected = addr in info_output and "Connected: yes" in info_output + devices.append({ + "address": addr, + "name": name, + "connected": connected, + "trusted": False, # Would need to check each device + }) + + return devices + + +def bt_pair(address: str, timeout: int = 30) -> dict[str, Any]: + """Pair with a Bluetooth device. + + Args: + address: Device MAC address (AA:BB:CC:DD:EE:FF). + timeout: Max seconds to wait for pairing. + + Returns: + dict with ok (bool) and error (str|None). + """ + if not address or ":" not in address: + return {"ok": False, "error": "Invalid MAC address"} + + address = address.upper() + + # Trust first (helps with pairing) + _bluetoothctl(f"trust {address}", timeout=10) + + # Attempt pairing + output = _bluetoothctl(f"pair {address}", timeout=timeout) + + if "Failed to pair" in output or "error" in output.lower(): + logger.warning("Pairing failed with %s", address) + return {"ok": False, "error": f"Pairing failed with {address}"} + + # Try to connect + _bluetoothctl(f"connect {address}", timeout=15) + time.sleep(1) + + logger.info("Paired with %s", address) + return {"ok": True} + + +def bt_unpair(address: str) -> dict[str, Any]: + """Unpair/remove a Bluetooth device. + + Args: + address: Device MAC address. + + Returns: + dict with ok (bool). + """ + if not address or ":" not in address: + return {"ok": False, "error": "Invalid MAC address"} + + address = address.upper() + + # Disconnect first + _bluetoothctl(f"disconnect {address}", timeout=10) + time.sleep(0.3) + + # Remove + output = _bluetoothctl(f"remove {address}", timeout=10) + success = "not available" not in output + + if success: + logger.info("Unpaired %s", address) + return {"ok": success} + + +def bt_connect(address: str) -> dict[str, Any]: + """Connect to a paired Bluetooth device. + + Args: + address: Device MAC address. + + Returns: + dict with ok (bool) and error (str|None). + """ + if not address or ":" not in address: + return {"ok": False, "error": "Invalid MAC address"} + + address = address.upper() + output = _bluetoothctl(f"connect {address}", timeout=20) + + if "Connection successful" in output or "successful" in output.lower(): + return {"ok": True} + else: + logger.warning("Connection to %s failed", address) + return {"ok": False, "error": "Connection failed"} + + +def bt_disconnect(address: str) -> dict[str, Any]: + """Disconnect a Bluetooth device. + + Args: + address: Device MAC address. + + Returns: + dict with ok (bool). + """ + if not address or ":" not in address: + return {"ok": False, "error": "Invalid MAC address"} + + address = address.upper() + _bluetoothctl(f"disconnect {address}", timeout=10) + return {"ok": True} + + +# ── Bluetooth MIDI service management ───────────────────────────────────────── + + +def bt_midi_status() -> dict[str, Any]: + """Check Bluetooth MIDI service status. + + Returns dict with: + enabled (bool): service is enabled (starts on boot). + running (bool): service is currently running. + """ + # Check if service is enabled + enabled = False + running = False + + try: + result = subprocess.run( + ["systemctl", "is-enabled", BT_MIDI_SERVICE_NAME], + capture_output=True, text=True, timeout=10, + ) + enabled = result.stdout.strip() == "enabled" + except Exception: + pass + + try: + result = subprocess.run( + ["systemctl", "is-active", BT_MIDI_SERVICE_NAME], + capture_output=True, text=True, timeout=10, + ) + running = result.stdout.strip() == "active" + except Exception: + pass + + return { + "enabled": enabled, + "running": running, + } + + +def bt_midi_enable() -> dict[str, Any]: + """Enable and start the Bluetooth MIDI service. + + The service must be installed first (via scripts/setup-bt-midi.sh). + + Returns dict with ok (bool) and error (str|None). + """ + try: + subprocess.run( + ["sudo", "systemctl", "enable", BT_MIDI_SERVICE_NAME], + capture_output=True, text=True, timeout=15, check=True, + ) + subprocess.run( + ["sudo", "systemctl", "start", BT_MIDI_SERVICE_NAME], + capture_output=True, text=True, timeout=15, check=True, + ) + logger.info("Bluetooth MIDI service enabled and started") + return {"ok": True} + except subprocess.CalledProcessError as e: + logger.warning("Failed to enable BT MIDI: %s", e.stderr if e.stderr else str(e)) + return {"ok": False, "error": e.stderr.strip() if e.stderr else str(e)} + + +def bt_midi_disable() -> dict[str, Any]: + """Stop and disable the Bluetooth MIDI service. + + Returns dict with ok (bool). + """ + try: + subprocess.run( + ["sudo", "systemctl", "stop", BT_MIDI_SERVICE_NAME], + capture_output=True, text=True, timeout=15, + ) + subprocess.run( + ["sudo", "systemctl", "disable", BT_MIDI_SERVICE_NAME], + capture_output=True, text=True, timeout=15, + ) + logger.info("Bluetooth MIDI service stopped and disabled") + return {"ok": True} + except Exception as e: + logger.warning("Failed to disable BT MIDI: %s", e) + return {"ok": False} + + +def bt_midi_connected_devices() -> list[dict[str, Any]]: + """List devices connected via Bluetooth MIDI. + + Checks which paired devices are currently connected and + might be MIDI controllers. + + Returns list of dicts with address, name. + """ + # Get the info for all paired devices + paired = bt_paired_devices() + # Filter to connected ones + midi_devices = [d for d in paired if d.get("connected")] + return midi_devices \ No newline at end of file diff --git a/src/system/network.py b/src/system/network.py new file mode 100644 index 0000000..435c268 --- /dev/null +++ b/src/system/network.py @@ -0,0 +1,620 @@ +"""WiFi network management module for Pi Multi-FX Pedal. + +Uses nmcli (NetworkManager) by default — works on RPi OS and most +desktop Linux distros. Falls back to iwlist + manual wpa_supplicant +for minimized DietPi installs without NetworkManager. + +All public methods return dicts suitable for JSON serialization. +""" + +from __future__ import annotations + +import json +import logging +import os +import re +import subprocess +import tempfile +import time +from pathlib import Path +from typing import Any, Optional + +logger = logging.getLogger(__name__) + +# ── Constants ───────────────────────────────────────────────────────────────── + +HOTSPOT_SCRIPT = Path(__file__).resolve().parent.parent.parent / "scripts" / "setup-wifi-ap.sh" +WPA_SUPPLICANT_CONF = Path("/etc/wpa_supplicant/wpa_supplicant.conf") + +# ── Capability detection ────────────────────────────────────────────────────── + + +def _has_nmcli() -> bool: + """Check if NetworkManager (nmcli) is available on this system.""" + try: + subprocess.run( + ["which", "nmcli"], + capture_output=True, timeout=5, check=True, + ) + return True + except (subprocess.CalledProcessError, FileNotFoundError, OSError): + return False + + +def _run(cmd: list[str], timeout: int = 15, check: bool = True) -> subprocess.CompletedProcess: + """Run a subprocess and return the result.""" + logger.debug("Running: %s", " ".join(cmd)) + return subprocess.run( + cmd, capture_output=True, text=True, timeout=timeout, check=check, + ) + + +# ── nmcli helpers ───────────────────────────────────────────────────────────── + + +def _nmcli_scan() -> list[dict[str, Any]]: + """Scan WiFi networks using nmcli. + + Returns a list of networks with SSID, signal strength (%), and encryption type. + """ + result = _run( + ["nmcli", "-t", "-f", "SSID,SIGNAL,SECURITY", "device", "wifi", "list"], + timeout=20, + check=False, + ) + networks: list[dict[str, Any]] = [] + seen: set[str] = set() + for line in result.stdout.strip().split("\n"): + if not line.strip() or ":" not in line: + continue + parts = line.split(":", 2) + if len(parts) < 3: + continue + ssid, signal_str, security = parts + ssid = ssid.strip() + if not ssid or ssid in seen: + continue + seen.add(ssid) + try: + signal = int(signal_str) + except ValueError: + signal = 0 + networks.append({ + "ssid": ssid, + "signal": signal, + "encryption": security.strip() if security.strip() and security.strip() != "" else "Open", + }) + return sorted(networks, key=lambda n: n["signal"], reverse=True) + + +def _nmcli_status() -> dict[str, Any]: + """Get current WiFi connection status via nmcli. + + Returns SSID, IP address, signal strength, and device name. + """ + result = _run( + ["nmcli", "-t", "-f", "DEVICE,TYPE,STATE,CONNECTION", "device", "status"], + timeout=10, check=False, + ) + wifi_device = None + for line in result.stdout.strip().split("\n"): + parts = line.split(":") + if len(parts) >= 3 and parts[1] == "wifi": + wifi_device = parts[0] + break + + if not wifi_device: + return {"connected": False, "ssid": None, "ip": None, "signal": 0} + + # Get connection details + detail = _run( + ["nmcli", "-t", "-f", "GENERAL.CONNECTION,IP4.ADDRESS", "device", "show", wifi_device], + timeout=10, check=False, + ) + ssid = None + ip = None + for line in detail.stdout.strip().split("\n"): + if line.startswith("GENERAL.CONNECTION:"): + val = line.split(":", 1)[1].strip() + if val and val != "--": + ssid = val + elif line.startswith("IP4.ADDRESS:"): + val = line.split(":", 1)[1].strip() + if val and val != "--": + ip = val.split("/")[0] + + if not ssid: + return {"connected": False, "ssid": None, "ip": None, "signal": 0} + + # Get signal strength + signal = 0 + sig_result = _run( + ["nmcli", "-t", "-f", "SSID,SIGNAL", "device", "wifi", "list"], + timeout=10, check=False, + ) + for line in sig_result.stdout.strip().split("\n"): + if ":" in line: + s, sig = line.split(":", 1) + if s.strip() == ssid: + try: + signal = int(sig) + except ValueError: + pass + break + + return { + "connected": True, + "ssid": ssid, + "ip": ip, + "signal": signal, + "device": wifi_device, + } + + +def _nmcli_saved() -> list[dict[str, Any]]: + """List saved/known WiFi connections.""" + result = _run( + ["nmcli", "-t", "-f", "NAME,UUID,TYPE", "connection", "show"], + timeout=10, check=False, + ) + saved: list[dict[str, Any]] = [] + for line in result.stdout.strip().split("\n"): + parts = line.split(":") + if len(parts) >= 3 and "wireless" in parts[2]: + saved.append({ + "ssid": parts[0], + "uuid": parts[1], + }) + return saved + + +def _nmcli_connect(ssid: str, password: str) -> bool: + """Connect to a WiFi network. + + If the network is already saved, just activate it. + Otherwise create a new connection profile. + """ + # Check if already saved + saved = _nmcli_saved() + for net in saved: + if net["ssid"] == ssid: + _run( + ["nmcli", "connection", "up", net["uuid"]], + timeout=30, check=False, + ) + logger.info("Activated saved network: %s", ssid) + return True + + # Connect with password + result = _run( + ["nmcli", "device", "wifi", "connect", ssid, "password", password], + timeout=30, check=False, + ) + ok = result.returncode == 0 + if ok: + logger.info("Connected to new network: %s", ssid) + else: + logger.warning("Failed to connect to %s: %s", ssid, result.stderr.strip()) + return ok + + +def _nmcli_disconnect() -> bool: + """Disconnect from the current WiFi network.""" + status = _nmcli_status() + if not status.get("connected"): + return True # already disconnected + device = status.get("device", "wlan0") + result = _run( + ["nmcli", "device", "disconnect", device], + timeout=15, check=False, + ) + return result.returncode == 0 + + +def _nmcli_forget(ssid: str) -> bool: + """Delete a saved WiFi network.""" + saved = _nmcli_saved() + for net in saved: + if net["ssid"] == ssid: + _run( + ["nmcli", "connection", "delete", net["uuid"]], + timeout=15, check=False, + ) + logger.info("Forgot network: %s", ssid) + return True + return False + + +# ── iwlist fallback (minimized DietPi) ──────────────────────────────────────── + + +def _iwlist_scan() -> list[dict[str, Any]]: + """Fallback WiFi scan using iwlist (no NetworkManager).""" + result = _run( + ["sudo", "iwlist", "wlan0", "scan"], + timeout=30, check=False, + ) + networks: list[dict[str, Any]] = [] + current: dict[str, Any] = {} + for line in result.stdout.split("\n"): + line = line.strip() + if "ESSID:" in line: + essid = line.split("ESSID:")[1].strip().strip('"') + if essid and essid != "": + current["ssid"] = essid + elif "Signal level=" in line: + match = re.search(r"Signal level=(-?\d+)", line) + if match: + dbm = int(match.group(1)) + # Convert dBm to percentage (0% = -100dBm, 100% = -30dBm) + signal = max(0, min(100, int((dbm + 100) * 100 / 70))) + current["signal"] = signal + elif "Encryption key:" in line: + val = line.split(":")[1].strip() + current["encryption"] = "WPA/WPA2" if val == "on" else "Open" + elif "IE:" in line and "WPA" in line.upper(): + if current.get("encryption") in (None, "Open"): + current["encryption"] = "WPA/WPA2" + if current.get("ssid") and "signal" in current and "encryption" in current: + networks.append(dict(current)) + current = {} + + seen: set[str] = set() + unique: list[dict[str, Any]] = [] + for n in networks: + if n["ssid"] not in seen: + seen.add(n["ssid"]) + unique.append(n) + return sorted(unique, key=lambda n: n.get("signal", 0), reverse=True) + + +def _iwlist_status() -> dict[str, Any]: + """Fallback WiFi status using iwconfig.""" + result = _run( + ["iwconfig", "wlan0"], + timeout=10, check=False, + ) + stdout = result.stdout + ssid = None + signal = 0 + match_ssid = re.search(r'ESSID:"([^"]+)"', stdout) + if match_ssid: + ssid = match_ssid.group(1) + match_signal = re.search(r"Signal level=(-?\d+)", stdout) + if match_signal: + dbm = int(match_signal.group(1)) + signal = max(0, min(100, int((dbm + 100) * 100 / 70))) + + # Get IP + ip = None + ip_result = _run( + ["ip", "-4", "addr", "show", "wlan0"], + timeout=5, check=False, + ) + ip_match = re.search(r"inet (\d+\.\d+\.\d+\.\d+)", ip_result.stdout) + if ip_match: + ip = ip_match.group(1) + + if not ssid or ssid == "off/any": + return {"connected": False, "ssid": None, "ip": None, "signal": 0} + return {"connected": True, "ssid": ssid, "ip": ip, "signal": signal, "device": "wlan0"} + + +def _wpa_connect(ssid: str, password: str) -> bool: + """Fallback: add network to wpa_supplicant.conf and restart.""" + try: + # Generate psk + psk_result = _run( + ["wpa_passphrase", ssid, password], + timeout=10, check=True, + ) + network_block = psk_result.stdout.strip() + + # Append to wpa_supplicant.conf + if WPA_SUPPLICANT_CONF.exists(): + current = WPA_SUPPLICANT_CONF.read_text() + if ssid in current: + # Remove old config for this SSID and re-add + lines = current.split("\n") + new_lines = [] + skip = False + for line in lines: + if f'ssid="{ssid}"' in line: + skip = True + continue + if skip and "}" in line: + skip = False + continue + if not skip: + new_lines.append(line) + current = "\n".join(new_lines) + + updated = current.strip() + "\n\n" + network_block + "\n" + else: + updated = network_block + "\n" + + WPA_SUPPLICANT_CONF.parent.mkdir(parents=True, exist_ok=True) + # Write via temp to avoid partial writes + with tempfile.NamedTemporaryFile( + mode="w", dir=str(WPA_SUPPLICANT_CONF.parent), + prefix="wpa_temp_", delete=False, + ) as f: + f.write(updated) + tmp_path = f.name + _run(["sudo", "cp", tmp_path, str(WPA_SUPPLICANT_CONF)], timeout=10) + os.unlink(tmp_path) + + # Restart networking + _run(["sudo", "wpa_cli", "-i", "wlan0", "reconfigure"], timeout=10, check=False) + time.sleep(3) + return True + except Exception as e: + logger.error("Failed to connect with wpa_supplicant: %s", e) + return False + + +# ── Hotspot control ─────────────────────────────────────────────────────────── + + +def _hotspot_status() -> dict[str, Any]: + """Check if the WiFi hotspot is currently active.""" + # Check if hostapd is running + result = _run( + ["systemctl", "is-active", "hostapd"], + timeout=10, check=False, + ) + active = result.stdout.strip() == "active" + + if not active: + return {"active": False, "ssid": None, "clients": 0, "ip": None} + + # Read SSID from config + ssid = None + hostapd_conf = Path("/etc/hostapd/hostapd.conf") + if hostapd_conf.exists(): + for line in hostapd_conf.read_text().split("\n"): + if line.startswith("ssid="): + ssid = line.split("=", 1)[1].strip() + break + + # Count connected clients via iw + clients = 0 + iw_result = _run( + ["iw", "dev", "wlan0", "station", "dump"], + timeout=10, check=False, + ) + if iw_result.returncode == 0: + clients = len([l for l in iw_result.stdout.split("\n") if "Station" in l]) + + return { + "active": True, + "ssid": ssid or "Pi-Pedal", + "clients": clients, + "ip": "192.168.4.1", + } + + +def _hotspot_enable(ssid: str = "Pi-Pedal", password: str = "pedal1234") -> bool: + """Enable WiFi hotspot mode by calling the setup script.""" + result = _run( + ["sudo", "bash", str(HOTSPOT_SCRIPT), "--ssid", ssid, "--psk", password], + timeout=60, check=False, + ) + ok = result.returncode == 0 + if ok: + logger.info("Hotspot enabled: SSID=%s", ssid) + else: + logger.warning("Failed to enable hotspot: %s", result.stderr.strip()) + return ok + + +def _hotspot_disable() -> bool: + """Disable WiFi hotspot, return to client mode.""" + result = _run( + ["sudo", "bash", str(HOTSPOT_SCRIPT), "--disable"], + timeout=60, check=False, + ) + ok = result.returncode == 0 + if ok: + logger.info("Hotspot disabled") + else: + logger.warning("Failed to disable hotspot: %s", result.stderr.strip()) + return ok + + +# ── Public API ───────────────────────────────────────────────────────────────── + + +def wifi_scan() -> list[dict[str, Any]]: + """Scan for available WiFi networks. + + Returns a list of dicts with keys: ssid, signal (%), encryption. + """ + if _has_nmcli(): + try: + return _nmcli_scan() + except Exception as e: + logger.warning("nmcli scan failed, falling back to iwlist: %s", e) + try: + return _iwlist_scan() + except Exception as e: + logger.error("All WiFi scan methods failed: %s", str(e)) + return [] + + +def wifi_status() -> dict[str, Any]: + """Get current WiFi client connection status. + + Returns dict with: connected (bool), ssid (str|None), ip (str|None), + signal (int), device (str|None). + """ + if _has_nmcli(): + try: + return _nmcli_status() + except Exception as e: + logger.warning("nmcli status failed, falling back: %s", e) + try: + return _iwlist_status() + except Exception as e: + logger.error("All WiFi status methods failed: %s", str(e)) + return {"connected": False, "ssid": None, "ip": None, "signal": 0} + + +def wifi_saved_networks() -> list[dict[str, Any]]: + """List saved/known WiFi networks. + + Returns list of dicts with keys: ssid, uuid (nmcli) or just ssid. + """ + if _has_nmcli(): + try: + return _nmcli_saved() + except Exception as e: + logger.warning("nmcli saved networks failed: %s", e) + # Fallback: parse wpa_supplicant.conf + try: + if WPA_SUPPLICANT_CONF.exists(): + networks: list[dict[str, Any]] = [] + for line in WPA_SUPPLICANT_CONF.read_text().split("\n"): + match = re.search(r'ssid="([^"]+)"', line) + if match: + networks.append({"ssid": match.group(1)}) + return networks + except Exception as e: + logger.warning("Failed to read wpa_supplicant.conf: %s", e) + return [] + + +def wifi_connect(ssid: str, password: str) -> dict[str, Any]: + """Connect to a WiFi network. + + Args: + ssid: Network SSID. + password: Network password. + + Returns: + dict with ok (bool) and error (str|None) keys. + """ + if not ssid: + return {"ok": False, "error": "SSID is required"} + if not password: + return {"ok": False, "error": "Password is required"} + + if _has_nmcli(): + try: + ok = _nmcli_connect(ssid, password) + return {"ok": ok, "error": None if ok else "Failed to connect via NetworkManager"} + except Exception as e: + logger.warning("nmcli connect failed, falling back: %s", e) + + try: + ok = _wpa_connect(ssid, password) + return {"ok": ok, "error": None if ok else "Failed to connect via wpa_supplicant"} + except Exception as e: + logger.error("All connect methods failed: %s", str(e)) + return {"ok": False, "error": str(e)} + + +def wifi_disconnect() -> dict[str, Any]: + """Disconnect from the current WiFi network. + + Returns dict with ok (bool) key. + """ + if _has_nmcli(): + try: + ok = _nmcli_disconnect() + return {"ok": ok} + except Exception as e: + logger.warning("nmcli disconnect failed: %s", e) + try: + _run(["sudo", "iwconfig", "wlan0", "essid", "off"], timeout=10, check=False) + return {"ok": True} + except Exception as e: + logger.error("Disconnect failed: %s", str(e)) + return {"ok": False} + + +def wifi_forget(ssid: str) -> dict[str, Any]: + """Forget a saved WiFi network. + + Args: + ssid: Network SSID to forget. + + Returns: + dict with ok (bool) key. + """ + if _has_nmcli(): + try: + ok = _nmcli_forget(ssid) + return {"ok": ok} + except Exception as e: + logger.warning("nmcli forget failed: %s", e) + # Fallback: remove from wpa_supplicant.conf + try: + if WPA_SUPPLICANT_CONF.exists(): + lines = WPA_SUPPLICANT_CONF.read_text().split("\n") + new_lines: list[str] = [] + skip = False + removed = False + for line in lines: + if f'ssid="{ssid}"' in line: + skip = True + removed = True + continue + if skip and "}" in line: + skip = False + continue + if not skip: + new_lines.append(line) + if removed: + WPA_SUPPLICANT_CONF.write_text("\n".join(new_lines)) + _run(["sudo", "wpa_cli", "-i", "wlan0", "reconfigure"], timeout=10, check=False) + return {"ok": True} + return {"ok": False} + except Exception as e: + logger.error("Forget failed: %s", str(e)) + return {"ok": False} + + +def hotspot_status() -> dict[str, Any]: + """Get the current hotspot status. + + Returns dict with active (bool), ssid (str|None), clients (int), ip (str|None). + """ + try: + return _hotspot_status() + except Exception as e: + logger.error("Hotspot status failed: %s", str(e)) + return {"active": False, "ssid": None, "clients": 0, "ip": None} + + +def hotspot_enable(ssid: str = "Pi-Pedal", password: str = "pedal1234") -> dict[str, Any]: + """Enable the WiFi hotspot. + + Args: + ssid: Hotspot SSID. + password: Hotspot password (min 8 chars). + + Returns: + dict with ok (bool) and error (str|None) keys. + """ + if len(password) < 8: + return {"ok": False, "error": "Password must be at least 8 characters"} + try: + ok = _hotspot_enable(ssid, password) + return {"ok": ok, "error": None if ok else "Failed to enable hotspot"} + except Exception as e: + logger.error("Hotspot enable failed: %s", str(e)) + return {"ok": False, "error": str(e)} + + +def hotspot_disable() -> dict[str, Any]: + """Disable the WiFi hotspot. + + Returns dict with ok (bool) key. + """ + try: + ok = _hotspot_disable() + return {"ok": ok} + except Exception as e: + logger.error("Hotspot disable failed: %s", str(e)) + return {"ok": False} \ No newline at end of file diff --git a/src/web/server.py b/src/web/server.py index b0cd228..f876714 100644 --- a/src/web/server.py +++ b/src/web/server.py @@ -540,6 +540,177 @@ class WebServer: logger.info("Uploaded IR file: %s", dest.name) return {"ok": True, "filename": dest.name} + # ── WiFi endpoints ────────────────────────────────────── + + @app.get("/api/wifi/status") + async def wifi_get_status(): + """Get current WiFi client connection status.""" + from ..system.network import wifi_status + return wifi_status() + + @app.get("/api/wifi/scan") + async def wifi_scan(): + """Scan for available WiFi networks.""" + from ..system.network import wifi_scan + return {"networks": wifi_scan()} + + @app.get("/api/wifi/saved") + async def wifi_saved(): + """List saved/known WiFi networks.""" + from ..system.network import wifi_saved_networks + return {"networks": wifi_saved_networks()} + + @app.post("/api/wifi/connect") + async def wifi_connect(data: dict): + """Connect to a WiFi network. + + Body: {ssid: str, password: str} + """ + from ..system.network import wifi_connect + return wifi_connect( + ssid=data.get("ssid", ""), + password=data.get("password", ""), + ) + + @app.post("/api/wifi/disconnect") + async def wifi_disconnect(): + """Disconnect from the current WiFi network.""" + from ..system.network import wifi_disconnect + return wifi_disconnect() + + @app.post("/api/wifi/forget") + async def wifi_forget(data: dict): + """Forget a saved WiFi network. + + Body: {ssid: str} + """ + from ..system.network import wifi_forget + return wifi_forget(ssid=data.get("ssid", "")) + + @app.get("/api/wifi/hotspot") + async def wifi_hotspot_status(): + """Get hotspot status.""" + from ..system.network import hotspot_status + return hotspot_status() + + @app.post("/api/wifi/hotspot/enable") + async def wifi_hotspot_enable(data: dict): + """Enable WiFi hotspot. + + Body: {ssid: str (optional), password: str (optional)} + """ + from ..system.network import hotspot_enable + return hotspot_enable( + ssid=data.get("ssid", "Pi-Pedal"), + password=data.get("password", "pedal1234"), + ) + + @app.post("/api/wifi/hotspot/disable") + async def wifi_hotspot_disable(): + """Disable WiFi hotspot.""" + from ..system.network import hotspot_disable + return hotspot_disable() + + # ── Bluetooth endpoints ────────────────────────────────── + + @app.get("/api/bluetooth/status") + async def bluetooth_status(): + """Get Bluetooth adapter status.""" + from ..system.bluetooth import bt_status + return bt_status() + + @app.post("/api/bluetooth/power") + async def bluetooth_power(data: dict): + """Power Bluetooth on or off. + + Body: {on: bool} + """ + from ..system.bluetooth import bt_power + return bt_power(on=data.get("on", True)) + + @app.post("/api/bluetooth/discoverable") + async def bluetooth_discoverable(data: dict): + """Set Bluetooth discoverable. + + Body: {on: bool} + """ + from ..system.bluetooth import bt_discoverable + return bt_discoverable(on=data.get("on", True)) + + @app.post("/api/bluetooth/scan") + async def bluetooth_scan(): + """Scan for discoverable Bluetooth devices.""" + from ..system.bluetooth import bt_scan + return bt_scan(timeout=12) + + @app.get("/api/bluetooth/paired") + async def bluetooth_paired(): + """List paired Bluetooth devices.""" + from ..system.bluetooth import bt_paired_devices + return {"devices": bt_paired_devices()} + + @app.post("/api/bluetooth/pair") + async def bluetooth_pair(data: dict): + """Pair with a Bluetooth device. + + Body: {address: str} + """ + from ..system.bluetooth import bt_pair + return bt_pair(address=data.get("address", "")) + + @app.post("/api/bluetooth/unpair") + async def bluetooth_unpair(data: dict): + """Unpair a Bluetooth device. + + Body: {address: str} + """ + from ..system.bluetooth import bt_unpair + return bt_unpair(address=data.get("address", "")) + + @app.post("/api/bluetooth/connect") + async def bluetooth_connect(data: dict): + """Connect to a paired Bluetooth device. + + Body: {address: str} + """ + from ..system.bluetooth import bt_connect + return bt_connect(address=data.get("address", "")) + + @app.post("/api/bluetooth/disconnect") + async def bluetooth_disconnect(data: dict): + """Disconnect a Bluetooth device. + + Body: {address: str} + """ + from ..system.bluetooth import bt_disconnect + return bt_disconnect(address=data.get("address", "")) + + @app.get("/api/bluetooth/midi-status") + async def bluetooth_midi_status(): + """Get Bluetooth MIDI service status.""" + from ..system.bluetooth import bt_midi_status + return bt_midi_status() + + @app.post("/api/bluetooth/midi-enable") + async def bluetooth_midi_enable(): + """Enable and start the Bluetooth MIDI service.""" + from ..system.bluetooth import bt_midi_enable + return bt_midi_enable() + + @app.post("/api/bluetooth/midi-disable") + async def bluetooth_midi_disable(): + """Stop and disable the Bluetooth MIDI service.""" + from ..system.bluetooth import bt_midi_disable + return bt_midi_disable() + + @app.get("/api/bluetooth/midi-devices") + async def bluetooth_midi_devices(): + """List connected Bluetooth MIDI devices.""" + from ..system.bluetooth import bt_midi_connected_devices + return {"devices": bt_midi_connected_devices()} + + # ── FX block param schemas ─────────────────────────────── + @app.get("/api/block-params/{fx_type}") async def get_block_params(fx_type: str): """Return editable parameters for a given FX block type. @@ -581,6 +752,47 @@ class WebServer: return app + # ── Network/BT state gathering ────────────────────────────────── + + def _gather_wifi_state(self) -> dict: + """Try to gather WiFi status without hard-failing.""" + try: + from ..system.network import wifi_status, hotspot_status + ws = wifi_status() + hs = hotspot_status() + return { + "connected": ws.get("connected", False), + "ssid": ws.get("ssid"), + "ip": ws.get("ip"), + "signal": ws.get("signal", 0), + "hotspot_active": hs.get("active", False), + "hotspot_ssid": hs.get("ssid"), + "hotspot_clients": hs.get("clients", 0), + } + except Exception: + return {"connected": False, "ssid": None, "ip": None, "signal": 0, + "hotspot_active": False, "hotspot_ssid": None, "hotspot_clients": 0} + + def _gather_bt_state(self) -> dict: + """Try to gather Bluetooth status without hard-failing.""" + try: + from ..system.bluetooth import bt_status, bt_midi_status + bs = bt_status() + ms = bt_midi_status() + return { + "available": bs.get("available", False), + "powered": bs.get("powered", False), + "name": bs.get("name"), + "address": bs.get("address"), + "discoverable": bs.get("discoverable", False), + "midi_enabled": ms.get("enabled", False), + "midi_running": ms.get("running", False), + } + except Exception: + return {"available": False, "powered": False, "name": None, + "address": None, "discoverable": False, + "midi_enabled": False, "midi_running": False} + # ── State gathering ────────────────────────────────────────────── def _gather_state(self) -> dict[str, Any]: @@ -631,6 +843,8 @@ class WebServer: "nam_model": current_model_name, "ir_loaded": bool(ir and ir.is_loaded) if ir else False, "ir_name": current_ir_name, + "wifi": self._gather_wifi_state(), + "bluetooth": self._gather_bt_state(), } # ── Lifecycle ──────────────────────────────────────────────────── diff --git a/src/web/static/bluetooth.js b/src/web/static/bluetooth.js new file mode 100644 index 0000000..a9b6948 --- /dev/null +++ b/src/web/static/bluetooth.js @@ -0,0 +1,294 @@ +/** + * bluetooth.js — Bluetooth scan, pair, MIDI management for Pi Multi-FX Pedal. + * + * Relies on apiGet, apiPost from app.js and the pedalWS singleton. + */ + +/* ── BT Status ─────────────────────────────────────────────────── */ + +async function refreshBTStatus() { + try { + const status = await apiGet('/bluetooth/status'); + const powerBtn = document.getElementById('bt-power-btn'); + const discBtn = document.getElementById('bt-discoverable-btn'); + const nameEl = document.getElementById('bt-adapter-name'); + const addrEl = document.getElementById('bt-address'); + + if (powerBtn) { + powerBtn.textContent = status.powered ? 'ON' : 'OFF'; + powerBtn.classList.toggle('active', status.powered); + } + if (discBtn) { + discBtn.textContent = status.discoverable ? 'ON' : 'OFF'; + discBtn.classList.toggle('active', status.discoverable); + } + if (nameEl) nameEl.textContent = status.name || '—'; + if (addrEl) addrEl.textContent = status.address || '—'; + } catch (e) { + console.warn('Failed to get BT status:', e); + } +} + +/* ── BT Power ──────────────────────────────────────────────────── */ + +async function toggleBTPower() { + const btn = document.getElementById('bt-power-btn'); + const current = btn.classList.contains('active'); + const newState = !current; + btn.disabled = true; + try { + await apiPost('/bluetooth/power', { on: newState }); + refreshBTStatus(); + } catch (e) { + console.warn('BT power toggle failed:', e); + } finally { + btn.disabled = false; + } +} + +/* ── BT Discoverable ───────────────────────────────────────────── */ + +async function toggleBTDiscoverable() { + const btn = document.getElementById('bt-discoverable-btn'); + const current = btn.classList.contains('active'); + const newState = !current; + btn.disabled = true; + try { + await apiPost('/bluetooth/discoverable', { on: newState }); + refreshBTStatus(); + } catch (e) { + console.warn('BT discoverable toggle failed:', e); + } finally { + btn.disabled = false; + } +} + +/* ── BT Scan ───────────────────────────────────────────────────── */ + +async function scanBT() { + const btn = document.getElementById('bt-scan-btn'); + const results = document.getElementById('bt-scan-results'); + btn.disabled = true; + btn.textContent = 'Scanning (12s)...'; + results.innerHTML = '
Scanning for Bluetooth devices... (12 seconds)
'; + + try { + const data = await apiPost('/bluetooth/scan'); + const devices = data.devices || []; + + if (devices.length === 0) { + results.innerHTML = '
No devices found. Make sure devices are discoverable.
'; + return; + } + + let html = '
'; + for (const dev of devices) { + const rssiStr = dev.rssi !== null ? `${dev.rssi} dBm` : '—'; + const strengthClass = dev.rssi !== null && dev.rssi > -60 ? 'signal-strong' + : dev.rssi !== null && dev.rssi > -80 ? 'signal-medium' + : 'signal-weak'; + html += ` +
+
🔵
+
+
${escapeHtml(dev.name)}
+
+ ${dev.address} + ${rssiStr} +
+
+ +
`; + } + html += '
'; + results.innerHTML = html; + } catch (e) { + console.warn('BT scan failed:', e); + results.innerHTML = '
Scan failed.
'; + } finally { + btn.disabled = false; + btn.textContent = 'Scan'; + } +} + +/* ── Pair / Unpair ─────────────────────────────────────────────── */ + +async function pairBT(address) { + try { + const result = await apiPost('/bluetooth/pair', { address }); + if (result.ok) { + loadPairedDevices(); + // Refresh scan results to show paired + // alert('Paired successfully!'); + } else { + alert('Pairing failed: ' + (result.error || 'Unknown error')); + } + } catch (e) { + alert('Network error: ' + e.message); + } +} + +async function unpairBT(address) { + if (!confirm(`Unpair ${address}?`)) return; + try { + await apiPost('/bluetooth/unpair', { address }); + loadPairedDevices(); + } catch (e) { + console.warn('Unpair failed:', e); + } +} + +async function connectBT(address) { + try { + const result = await apiPost('/bluetooth/connect', { address }); + if (result.ok) { + loadPairedDevices(); + } else { + alert('Connection failed: ' + (result.error || 'Unknown error')); + } + } catch (e) { + console.warn('BT connect failed:', e); + } +} + +async function disconnectBT(address) { + try { + await apiPost('/bluetooth/disconnect', { address }); + loadPairedDevices(); + } catch (e) { + console.warn('BT disconnect failed:', e); + } +} + +/* ── Paired Devices ────────────────────────────────────────────── */ + +async function loadPairedDevices() { + const list = document.getElementById('bt-paired-list'); + try { + const data = await apiGet('/bluetooth/paired'); + const devices = data.devices || []; + + if (devices.length === 0) { + list.innerHTML = '
No paired devices.
'; + return; + } + + let html = '
'; + for (const dev of devices) { + const statusClass = dev.connected ? 'connected' : 'disconnected'; + const statusText = dev.connected ? 'Connected' : 'Disconnected'; + html += ` +
+
${dev.connected ? '🔗' : '🔘'}
+
+
${escapeHtml(dev.name)}
+
+ ${dev.address} + ${statusText} +
+
+
+ ${dev.connected + ? `` + : `` + } + +
+
`; + } + html += '
'; + list.innerHTML = html; + } catch (e) { + console.warn('Failed to load paired devices:', e); + list.innerHTML = '
Failed to load.
'; + } +} + +/* ── BT MIDI Service ───────────────────────────────────────────── */ + +async function refreshBTMIDIStatus() { + try { + const status = await apiGet('/bluetooth/midi-status'); + const indicator = document.getElementById('bt-midi-indicator'); + const enableBtn = document.getElementById('bt-midi-enable-btn'); + const disableBtn = document.getElementById('bt-midi-disable-btn'); + + if (indicator) { + if (status.running) { + indicator.textContent = 'Running'; + indicator.className = 'bt-midi-indicator bt-midi-on'; + } else if (status.enabled) { + indicator.textContent = 'Enabled (not running)'; + indicator.className = 'bt-midi-indicator bt-midi-warn'; + } else { + indicator.textContent = 'Off'; + indicator.className = 'bt-midi-indicator bt-midi-off'; + } + } + + // Load MIDI devices + try { + const devData = await apiGet('/bluetooth/midi-devices'); + const devEl = document.getElementById('bt-midi-devices'); + if (devEl) { + const devs = devData.devices || []; + if (devs.length > 0) { + devEl.innerHTML = devs.map(d => + `
${escapeHtml(d.name)} ${d.address}
` + ).join(''); + } else { + devEl.innerHTML = 'None connected'; + } + } + } catch (e) { + // MIDI devices are optional + } + } catch (e) { + console.warn('Failed to get BT MIDI status:', e); + } +} + +async function enableBTMIDI() { + const btn = document.getElementById('bt-midi-enable-btn'); + btn.disabled = true; + btn.textContent = 'Enabling...'; + try { + await apiPost('/bluetooth/midi-enable'); + refreshBTMIDIStatus(); + } catch (e) { + alert('Failed: ' + e.message); + } finally { + btn.disabled = false; + btn.textContent = 'Enable'; + } +} + +async function disableBTMIDI() { + const btn = document.getElementById('bt-midi-disable-btn'); + btn.disabled = true; + btn.textContent = 'Disabling...'; + try { + await apiPost('/bluetooth/midi-disable'); + refreshBTMIDIStatus(); + } catch (e) { + alert('Failed: ' + e.message); + } finally { + btn.disabled = false; + btn.textContent = 'Disable'; + } +} + +/* ── Helpers ────────────────────────────────────────────────────── */ + +function escapeHtml(str) { + const div = document.createElement('div'); + div.textContent = str; + return div.innerHTML; +} + +/* ── WebSocket listeners ────────────────────────────────────────── */ + +pedalWS.on('bluetooth_changed', (msg) => { + refreshBTStatus(); + refreshBTMIDIStatus(); +}); \ No newline at end of file diff --git a/src/web/static/network.js b/src/web/static/network.js new file mode 100644 index 0000000..45cb238 --- /dev/null +++ b/src/web/static/network.js @@ -0,0 +1,259 @@ +/** + * network.js — WiFi scan, connect, hotspot management for Pi Multi-FX Pedal. + * + * Relies on apiGet, apiPost from app.js and the pedalWS singleton. + */ + +/* ── WiFi Status ────────────────────────────────────────────────── */ + +async function refreshWiFiStatus() { + try { + const status = await apiGet('/wifi/status'); + const ssidEl = document.getElementById('wifi-ssid'); + const ipEl = document.getElementById('wifi-ip'); + const sigEl = document.getElementById('wifi-signal-text'); + const iconEl = document.getElementById('wifi-icon'); + + if (status.connected && status.ssid) { + ssidEl.textContent = status.ssid; + ipEl.textContent = status.ip || 'No IP'; + sigEl.textContent = `${status.signal || 0}%`; + // Signal bars + const sig = status.signal || 0; + if (sig > 70) iconEl.textContent = '📶'; + else if (sig > 40) iconEl.textContent = '📶'; + else if (sig > 15) iconEl.textContent = '📶'; + else iconEl.textContent = '📶'; + } else { + ssidEl.textContent = 'Not connected'; + ipEl.textContent = '—'; + sigEl.textContent = '—'; + iconEl.textContent = '📡'; + } + } catch (e) { + console.warn('Failed to get WiFi status:', e); + document.getElementById('wifi-ssid').textContent = 'Error'; + } +} + +/* ── WiFi Scan ──────────────────────────────────────────────────── */ + +async function scanWiFi() { + const btn = document.getElementById('wifi-scan-btn'); + const results = document.getElementById('wifi-scan-results'); + btn.disabled = true; + btn.textContent = 'Scanning...'; + results.innerHTML = '
Scanning for networks...
'; + + try { + const data = await apiGet('/wifi/scan'); + const networks = data.networks || []; + + if (networks.length === 0) { + results.innerHTML = '
No networks found. Make sure WiFi is enabled.
'; + return; + } + + let html = '
'; + for (const net of networks) { + const bars = signalBars(net.signal); + const encClass = net.encryption === 'Open' ? 'enc-open' : 'enc-secure'; + const displayEnc = net.encryption === 'Open' ? 'Open' : 'Secured'; + html += ` +
+
${bars}
+
+
${escapeHtml(net.ssid)}
+
${displayEnc} · ${net.signal}%
+
+ +
`; + } + html += '
'; + results.innerHTML = html; + } catch (e) { + console.warn('WiFi scan failed:', e); + results.innerHTML = '
Scan failed. Check WiFi hardware.
'; + } finally { + btn.disabled = false; + btn.textContent = 'Scan'; + } +} + +function signalBars(signal) { + if (signal > 70) return '📶'; + if (signal > 40) return '📶'; + if (signal > 15) return '📶'; + return '📶'; +} + +/* ── WiFi Connect Modal ─────────────────────────────────────────── */ + +let pendingSSID = ''; + +function openWiFiConnect(ssid) { + pendingSSID = ssid; + document.getElementById('wifi-modal-ssid').textContent = ssid; + document.getElementById('wifi-modal-password').value = ''; + document.getElementById('wifi-modal-error').textContent = ''; + document.getElementById('wifi-connect-modal').style.display = 'flex'; +} + +function closeWiFiModal() { + document.getElementById('wifi-connect-modal').style.display = 'none'; + pendingSSID = ''; +} + +async function connectWiFi() { + const password = document.getElementById('wifi-modal-password').value; + const errorEl = document.getElementById('wifi-modal-error'); + + if (!password) { + errorEl.textContent = 'Please enter a password.'; + return; + } + + errorEl.textContent = 'Connecting...'; + try { + const result = await apiPost('/wifi/connect', { ssid: pendingSSID, password }); + if (result.ok) { + errorEl.textContent = 'Connected!'; + setTimeout(() => { + closeWiFiModal(); + refreshWiFiStatus(); + }, 1500); + } else { + errorEl.textContent = result.error || 'Connection failed.'; + } + } catch (e) { + errorEl.textContent = 'Network error: ' + (e.message || 'Unknown'); + } +} + +/* ── Saved Networks ─────────────────────────────────────────────── */ + +async function loadSavedNetworks() { + const list = document.getElementById('wifi-saved-list'); + try { + const data = await apiGet('/wifi/saved'); + const networks = data.networks || []; + + if (networks.length === 0) { + list.innerHTML = '
No saved networks.
'; + return; + } + + let html = '
'; + for (const net of networks) { + html += ` +
+
+
${escapeHtml(net.ssid)}
+
Saved
+
+ +
`; + } + html += '
'; + list.innerHTML = html; + } catch (e) { + console.warn('Failed to load saved networks:', e); + list.innerHTML = '
Failed to load.
'; + } +} + +async function forgetNetwork(ssid) { + if (!confirm(`Forget "${ssid}"?`)) return; + try { + await apiPost('/wifi/forget', { ssid }); + loadSavedNetworks(); + refreshWiFiStatus(); + } catch (e) { + console.warn('Failed to forget network:', e); + } +} + +/* ── Hotspot ────────────────────────────────────────────────────── */ + +async function enableHotspot() { + const ssid = document.getElementById('hotspot-ssid').value.trim(); + const psk = document.getElementById('hotspot-psk').value.trim(); + const errorEl = document.getElementById('wifi-modal-error'); + + if (!ssid) { alert('SSID is required.'); return; } + if (psk.length < 8) { alert('Password must be at least 8 characters.'); return; } + + const btn = document.getElementById('hotspot-enable-btn'); + btn.disabled = true; + btn.textContent = 'Enabling...'; + + try { + const result = await apiPost('/wifi/hotspot/enable', { ssid, password: psk }); + if (result.ok) { + refreshWiFiStatus(); + updateHotspotUI(true, ssid); + } else { + alert('Failed: ' + (result.error || 'Unknown error')); + } + } catch (e) { + alert('Network error: ' + e.message); + } finally { + btn.disabled = false; + btn.textContent = 'Enable Hotspot'; + } +} + +async function disableHotspot() { + const btn = document.getElementById('hotspot-disable-btn'); + btn.disabled = true; + btn.textContent = 'Disabling...'; + + try { + const result = await apiPost('/wifi/hotspot/disable'); + if (result.ok) { + refreshWiFiStatus(); + updateHotspotUI(false); + } + } catch (e) { + alert('Network error: ' + e.message); + } finally { + btn.disabled = false; + btn.textContent = 'Disable Hotspot'; + } +} + +function updateHotspotUI(active, ssid) { + const indicator = document.getElementById('hotspot-status-indicator'); + const clients = document.getElementById('hotspot-clients'); + if (active) { + indicator.textContent = 'Active'; + indicator.className = 'hotspot-indicator hotspot-on'; + document.getElementById('hotspot-ssid-row').querySelector('input').value = ssid || 'Pi-Pedal'; + } else { + indicator.textContent = 'Off'; + indicator.className = 'hotspot-indicator hotspot-off'; + clients.textContent = '0'; + } +} + +/* ── Helpers ────────────────────────────────────────────────────── */ + +function escapeHtml(str) { + const div = document.createElement('div'); + div.textContent = str; + return div.innerHTML; +} + +/* ── WebSocket listeners ────────────────────────────────────────── */ + +pedalWS.on('network_changed', (msg) => { + refreshWiFiStatus(); +}); + +/* ── Initial load ───────────────────────────────────────────────── */ + +// Auto-load when the network tab is first activated (handled by switchTab) +// But also check if we're already on the settings page +if (document.getElementById('tab-network')) { + // Defer to DOMContentLoaded in settings.html → switchTab call +} \ No newline at end of file diff --git a/src/web/static/style.css b/src/web/static/style.css index 2e84156..a964bef 100644 --- a/src/web/static/style.css +++ b/src/web/static/style.css @@ -694,4 +694,304 @@ body { font-size: 0.78rem; color: var(--text-secondary); line-height: 1.4; +} + +/* ── Settings tabs ────────────────────────────────────────────────── */ + +.settings-tabs { + display: flex; + gap: 4px; + overflow-x: auto; + -webkit-overflow-scrolling: touch; + scrollbar-width: none; + margin-bottom: 12px; + padding-bottom: 4px; + border-bottom: 1px solid var(--border); +} + +.settings-tabs::-webkit-scrollbar { + display: none; +} + +.settings-tab { + padding: 8px 14px; + border: none; + background: transparent; + color: var(--text-secondary); + font-size: 0.82rem; + cursor: pointer; + white-space: nowrap; + border-bottom: 2px solid transparent; + transition: color 0.2s, border-color 0.2s; + touch-action: manipulation; +} + +.settings-tab:hover { + color: var(--text-primary); +} + +.settings-tab.active { + color: var(--accent); + border-bottom-color: var(--accent); +} + +.settings-panel { + display: none; +} + +.settings-panel.active { + display: block; +} + +/* ── Network UI ──────────────────────────────────────────────────── */ + +.network-status-row { + display: flex; + align-items: center; + gap: 12px; +} + +.network-status-icon { + font-size: 1.8rem; + width: 48px; + text-align: center; +} + +.network-status-info { + flex: 1; +} + +.network-ssid { + font-weight: 600; + font-size: 0.95rem; + margin-bottom: 2px; +} + +.network-ip { + font-size: 0.78rem; + color: var(--text-muted); +} + +.network-signal { + font-size: 0.78rem; + color: var(--text-secondary); +} + +.network-list { + display: flex; + flex-direction: column; + gap: 6px; +} + +.network-item { + display: flex; + align-items: center; + gap: 8px; + padding: 8px 10px; + background: var(--bg-secondary); + border: 1px solid var(--border); + border-radius: var(--radius-sm); +} + +.network-item-icon { + font-size: 1.2rem; + width: 28px; + text-align: center; +} + +.network-item-info { + flex: 1; + min-width: 0; +} + +.network-item-ssid { + font-size: 0.85rem; + font-weight: 500; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.network-item-meta { + font-size: 0.72rem; + color: var(--text-muted); + display: flex; + gap: 6px; +} + +.enc-open { + color: var(--success); +} + +.enc-secure { + color: var(--warning); +} + +.network-input { + background: var(--bg-secondary); + border: 1px solid var(--border); + border-radius: var(--radius-sm); + color: var(--text-primary); + padding: 6px 10px; + font-size: 0.85rem; + width: 100%; + max-width: 240px; + outline: none; +} + +.network-input:focus { + border-color: var(--accent); +} + +.network-error { + color: var(--danger); + font-size: 0.82rem; +} + +.btn-sm { + padding: 4px 10px; + font-size: 0.75rem; +} + +/* ── Hotspot ──────────────────────────────────────────────────────── */ + +.hotspot-controls .setting-row:last-child { + border-bottom: 1px solid var(--border); +} + +.hotspot-indicator { + display: inline-block; + padding: 2px 10px; + border-radius: 10px; + font-size: 0.78rem; + font-weight: 600; +} + +.hotspot-off { + background: var(--text-muted); + color: var(--bg-primary); +} + +.hotspot-on { + background: var(--warning); + color: #1a1a1a; +} + +.hotspot-actions { + display: flex; + gap: 8px; + padding-top: 10px; +} + +/* ── Bluetooth UI ────────────────────────────────────────────────── */ + +.bt-device-list { + display: flex; + flex-direction: column; + gap: 6px; +} + +.bt-device-item { + display: flex; + align-items: center; + gap: 8px; + padding: 8px 10px; + background: var(--bg-secondary); + border: 1px solid var(--border); + border-radius: var(--radius-sm); +} + +.bt-device-item.bt-paired { + border-left: 3px solid var(--text-muted); +} + +.bt-device-item.bt-paired.connected { + border-left-color: var(--success); +} + +.bt-device-icon { + font-size: 1.1rem; + width: 24px; + text-align: center; +} + +.bt-device-info { + flex: 1; + min-width: 0; +} + +.bt-device-name { + font-size: 0.85rem; + font-weight: 500; +} + +.bt-device-meta { + font-size: 0.72rem; + color: var(--text-muted); + display: flex; + gap: 6px; + align-items: center; +} + +.bt-device-meta code { + font-size: 0.68rem; + background: var(--bg-card); + padding: 1px 4px; + border-radius: 3px; +} + +.bt-device-actions { + display: flex; + gap: 4px; +} + +.bt-status-connected { + color: var(--success); + font-weight: 500; +} + +.bt-status-disconnected { + color: var(--text-muted); +} + +.signal-strong { color: var(--success); } +.signal-medium { color: var(--warning); } +.signal-weak { color: var(--text-muted); } + +.bt-midi-indicator { + display: inline-block; + padding: 2px 10px; + border-radius: 10px; + font-size: 0.78rem; + font-weight: 600; +} + +.bt-midi-on { + background: var(--success); + color: white; +} + +.bt-midi-warn { + background: var(--warning); + color: #1a1a1a; +} + +.bt-midi-off { + background: var(--text-muted); + color: var(--bg-primary); +} + +.bt-midi-actions { + display: flex; + gap: 8px; + padding: 8px 0; +} + +.bt-midi-device { + font-size: 0.82rem; + padding: 2px 0; +} + +.bt-midi-device code { + font-size: 0.72rem; } \ No newline at end of file diff --git a/src/web/templates/settings.html b/src/web/templates/settings.html index e20cc5e..a12c8d6 100644 --- a/src/web/templates/settings.html +++ b/src/web/templates/settings.html @@ -6,103 +6,318 @@

Settings

-
-
-

Connection

-
-
-
-
Hostname
-
pedal.local
-
-
-
Web Server
-
http://pedal.local:8080
-
-
-
Connected Clients
-
0
-
-
+ +
+ + + + + +
-
-
-

Audio

-
-
-
-
Master Volume
-
- - {{ (master_volume * 100)|int }}% + +
+
+

Connection

+
+
+
Hostname
+
pedal.local
-
-
-
Tuner
-
- +
+
Web Server
+
http://pedal.local:8080
+
+
+
Connected Clients
+
0
-
-
-

4-Cable Method (4CM)

-
-
-
-
4CM Mode
-
- + +
+
+

Audio

+
+
+
Master Volume
+
+ + {{ (master_volume * 100)|int }}% +
-
-
-
Breakpoint
-
- - {{ routing_breakpoint }} -
-
-
-
Routing
-
- - {% if routing_mode == '4cm' %} - Input 1 (Guitar) → Pre blocks [0..{{ routing_breakpoint }}) → Send -  |  Input 2 (Return) → Post blocks [{{ routing_breakpoint }}..] → Output - {% else %} - Mono: Guitar → Full chain → Output - {% endif %} - +
+
Tuner
+
+ +
-
-
-

About

+ +
+
+

4-Cable Method (4CM)

+
+
+
4CM Mode
+
+ +
+
+
+
Breakpoint
+
+ + {{ routing_breakpoint }} +
+
+
+
Routing
+
+ + {% if routing_mode == '4cm' %} + Input 1 (Guitar) → Pre blocks [0..{{ routing_breakpoint }}) → Send +  |  Input 2 (Return) → Post blocks [{{ routing_breakpoint }}..] → Output + {% else %} + Mono: Guitar → Full chain → Output + {% endif %} + +
+
+
-
-
-

Pi Multi-FX Pedal v0.1.0

-

Raspberry Pi 4B — Python 3.11 — JACK Audio

-

Gitea Repository

+
+ + +
+ +
+

WiFi Status

+
+
+
📶
+
+
—
+
—
+
—
+
+ +
+
+
+ + +
+
+

WiFi Networks

+ +
+
+
Click Scan to find available WiFi networks.
+
+
+ + +
+

Known Networks

+
+
Loading...
+
+
+ + +
+

WiFi Hotspot

+
+
+
+
Hotspot Status
+
+ Off +
+
+
+
SSID
+
+ +
+
+
+
Password
+
+ +
+
+
+
Connected Clients
+
0
+
+
+ + +
+
+
+
+
+ + +
+ +
+

Bluetooth

+
+
+
Bluetooth
+
+ +
+
+
+
Discoverable
+
+ +
+
+
+
Adapter Name
+
—
+
+
+
Address
+
—
+
+
+
+ + +
+
+

Discover Devices

+ +
+
+
Click Scan to find discoverable Bluetooth devices.
+
+
+ + +
+

Paired Devices

+
+
Loading...
+
+
+ + +
+

Bluetooth MIDI

+
+
+
BT MIDI Service
+
+ Off +
+
+
+ + +
+
+
Connected MIDI Devices
+
+ None connected +
+
+
+
+
+ + +
+
+

About

+
+
+

Pi Multi-FX Pedal v0.1.0

+

Raspberry Pi 4B — Python 3.11 — JACK Audio

+

Gitea Repository

+
+
+
+
+ + +
+{% endblock %} + +{% block scripts %} + + + {% endblock %} \ No newline at end of file diff --git a/tests/test_bt.py b/tests/test_bt.py new file mode 100644 index 0000000..464ef60 --- /dev/null +++ b/tests/test_bt.py @@ -0,0 +1,383 @@ +"""Tests for the Bluetooth management module (src.system.bluetooth). + +Uses mocking to avoid requiring actual Bluetooth hardware. + +Run with: + pytest tests/test_bt.py -v +""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +# Ensure src is on sys.path +SRC = Path(__file__).resolve().parent.parent / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from src.system.bluetooth import ( + bt_status, + bt_power, + bt_discoverable, + bt_scan, + bt_paired_devices, + bt_pair, + bt_unpair, + bt_connect, + bt_disconnect, + bt_midi_status, + bt_midi_enable, + bt_midi_disable, + bt_midi_connected_devices, +) + + +# ── Fixtures ───────────────────────────────────────────────────────────────── + + +@pytest.fixture +def mock_subprocess(): + """Patch subprocess.run globally for all BT tests.""" + with patch("src.system.bluetooth.subprocess.run") as mock: + yield mock + + +# ── bt_status() tests ───────────────────────────────────────────────────────── + + +def test_bt_status_powered_on(mock_subprocess): + """bt_status returns powered-on state.""" + mock_subprocess.return_value = MagicMock( + stdout="Controller AA:BB:CC:DD:EE:FF\n" + "\tName: pi-pedal\n" + "\tPowered: yes\n" + "\tDiscoverable: yes\n" + "\tPairable: yes\n", + returncode=0, + ) + status = bt_status() + assert status["available"] is True + assert status["powered"] is True + assert status["name"] == "pi-pedal" + assert status["discoverable"] is True + assert status["pairable"] is True + assert status["address"] == "AA:BB:CC:DD:EE:FF" + + +def test_bt_status_powered_off(mock_subprocess): + """bt_status returns powered-off state.""" + mock_subprocess.return_value = MagicMock( + stdout="Controller AA:BB:CC:DD:EE:FF\n" + "\tName: pi-pedal\n" + "\tPowered: no\n" + "\tDiscoverable: no\n" + "\tPairable: no\n", + returncode=0, + ) + status = bt_status() + assert status["powered"] is False + assert status["discoverable"] is False + + +def test_bt_status_not_available(mock_subprocess): + """bt_status returns not available when bluetoothctl fails.""" + mock_subprocess.side_effect = FileNotFoundError("no bluetoothctl") + status = bt_status() + assert status["available"] is False + assert status["powered"] is False + + +# ── bt_power() tests ────────────────────────────────────────────────────────── + + +def test_bt_power_on(mock_subprocess): + """bt_power(on=True) powers Bluetooth on.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="", returncode=0), # bluetoothctl power on + MagicMock( # bt_status call + stdout="Controller A\n\tPowered: yes\n", + returncode=0, + ), + ] + result = bt_power(on=True) + assert result["ok"] is True + + +def test_bt_power_off(mock_subprocess): + """bt_power(on=False) powers Bluetooth off.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="", returncode=0), # bluetoothctl power off + MagicMock( # bt_status call + stdout="Controller A\n\tPowered: no\n", + returncode=0, + ), + ] + result = bt_power(on=False) + assert result["ok"] is True + + +# ── bt_discoverable() tests ──────────────────────────────────────────────────── + + +def test_bt_discoverable_on(mock_subprocess): + """bt_discoverable(on=True) makes adapter discoverable.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="", returncode=0), # discoverable on + MagicMock( # bt_status + stdout="Controller A\n\tDiscoverable: yes\n", + returncode=0, + ), + ] + result = bt_discoverable(on=True) + assert result["ok"] is True + + +def test_bt_discoverable_off(mock_subprocess): + """bt_discoverable(on=False) hides adapter.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="", returncode=0), + MagicMock( + stdout="Controller A\n\tDiscoverable: no\n", + returncode=0, + ), + ] + result = bt_discoverable(on=False) + assert result["ok"] is True + + +# ── bt_scan() tests ─────────────────────────────────────────────────────────── + + +def test_bt_scan_finds_devices(mock_subprocess): + """bt_scan returns discovered devices.""" + # Simulate the three bluetoothctl calls + mock_subprocess.side_effect = [ + MagicMock(stdout="", returncode=0), # scan on + MagicMock(stdout="", returncode=0), # scan off + MagicMock( # devices + stdout="Device AA:BB:CC:DD:EE:FF My Guitar\n" + "Device 11:22:33:44:55:66 Expression Pedal\n", + returncode=0, + ), + ] + # bt_scan uses time.sleep which we don't want to actually wait + with patch("src.system.bluetooth.time.sleep", return_value=None): + result = bt_scan(timeout=1) + assert result["ok"] is True + assert len(result["devices"]) == 2 + assert result["devices"][0]["name"] == "My Guitar" + assert result["devices"][0]["address"] == "AA:BB:CC:DD:EE:FF" + assert result["devices"][1]["name"] == "Expression Pedal" + + +def test_bt_scan_no_devices(mock_subprocess): + """bt_scan returns empty list when no devices found.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="", returncode=0), # scan on + MagicMock(stdout="", returncode=0), # scan off + MagicMock(stdout="", returncode=0), # devices - empty + ] + with patch("src.system.bluetooth.time.sleep", return_value=None): + result = bt_scan(timeout=1) + assert result["ok"] is True + assert result["devices"] == [] + + +# ── bt_paired_devices() tests ────────────────────────────────────────────────── + + +def test_bt_paired_devices(mock_subprocess): + """bt_paired_devices returns paired device list.""" + mock_subprocess.side_effect = [ + MagicMock( # paired-devices + stdout="Device AA:BB:CC:DD:EE:FF MIDI Controller\n" + "Device 11:22:33:44:55:66 Footswitch\n", + returncode=0, + ), + MagicMock( # info + stdout="", + returncode=0, + ), + ] + devices = bt_paired_devices() + assert len(devices) == 2 + assert devices[0]["name"] == "MIDI Controller" + assert devices[0]["address"] == "AA:BB:CC:DD:EE:FF" + + +def test_bt_paired_devices_empty(mock_subprocess): + """bt_paired_devices returns empty list.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="", returncode=0), # paired-devices - empty + MagicMock(stdout="", returncode=0), + ] + devices = bt_paired_devices() + assert devices == [] + + +# ── bt_pair() tests ─────────────────────────────────────────────────────────── + + +def test_bt_pair_success(mock_subprocess): + """bt_pair successfully pairs with a device.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="", returncode=0), # trust + MagicMock(stdout="Pairing successful\n", returncode=0), # pair + MagicMock(stdout="", returncode=0), # connect + ] + with patch("src.system.bluetooth.time.sleep", return_value=None): + result = bt_pair("AA:BB:CC:DD:EE:FF") + assert result["ok"] is True + + +def test_bt_pair_invalid_address(): + """bt_pair returns error for invalid MAC address.""" + result = bt_pair("invalid") + assert result["ok"] is False + assert "Invalid" in result["error"] + + +def test_bt_pair_failure(mock_subprocess): + """bt_pair returns error when pairing fails.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="", returncode=0), # trust + MagicMock(stdout="Failed to pair\n", returncode=0), # pair + ] + with patch("src.system.bluetooth.time.sleep", return_value=None): + result = bt_pair("AA:BB:CC:DD:EE:FF") + assert result["ok"] is False + assert "failed" in result["error"].lower() + + +# ── bt_unpair() tests ────────────────────────────────────────────────────────── + + +def test_bt_unpair_success(mock_subprocess): + """bt_unpair removes a device.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="", returncode=0), # disconnect + MagicMock(stdout="Device has been removed\n", returncode=0), # remove + ] + with patch("src.system.bluetooth.time.sleep", return_value=None): + result = bt_unpair("AA:BB:CC:DD:EE:FF") + assert result["ok"] is True + + +def test_bt_unpair_invalid(): + """bt_unpair returns error for invalid address.""" + result = bt_unpair("invalid") + assert result["ok"] is False + + +# ── bt_connect() / bt_disconnect() tests ─────────────────────────────────────── + + +def test_bt_connect_success(mock_subprocess): + """bt_connect connects to a paired device.""" + mock_subprocess.return_value = MagicMock( + stdout="Connection successful\n", + returncode=0, + ) + result = bt_connect("AA:BB:CC:DD:EE:FF") + assert result["ok"] is True + + +def test_bt_connect_invalid(): + """bt_connect returns error for invalid address.""" + result = bt_connect("bad") + assert result["ok"] is False + + +def test_bt_disconnect_success(mock_subprocess): + """bt_disconnect disconnects a device.""" + mock_subprocess.return_value = MagicMock(stdout="", returncode=0) + result = bt_disconnect("AA:BB:CC:DD:EE:FF") + assert result["ok"] is True + + +def test_bt_disconnect_invalid(): + """bt_disconnect returns error for invalid address.""" + result = bt_disconnect("bad") + assert result["ok"] is False + + +# ── bt_midi_status() tests ───────────────────────────────────────────────────── + + +def test_bt_midi_status_enabled_running(mock_subprocess): + """bt_midi_status shows enabled and running.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="enabled\n", returncode=0), # is-enabled + MagicMock(stdout="active\n", returncode=0), # is-active + ] + status = bt_midi_status() + assert status["enabled"] is True + assert status["running"] is True + + +def test_bt_midi_status_disabled(mock_subprocess): + """bt_midi_status shows disabled.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="disabled\n", returncode=0), + MagicMock(stdout="inactive\n", returncode=0), + ] + status = bt_midi_status() + assert status["enabled"] is False + assert status["running"] is False + + +# ── bt_midi_enable() / bt_midi_disable() tests ───────────────────────────────── + + +def test_bt_midi_enable_success(mock_subprocess): + """bt_midi_enable starts the MIDI service.""" + mock_subprocess.return_value = MagicMock(returncode=0) + result = bt_midi_enable() + assert result["ok"] is True + + +def test_bt_midi_enable_failure(mock_subprocess): + """bt_midi_enable returns error on failure.""" + mock_subprocess.side_effect = subprocess.CalledProcessError(1, "systemctl") + result = bt_midi_enable() + assert result["ok"] is False + + +def test_bt_midi_disable_success(mock_subprocess): + """bt_midi_disable stops the MIDI service.""" + mock_subprocess.return_value = MagicMock(returncode=0) + result = bt_midi_disable() + assert result["ok"] is True + + +# ── bt_midi_connected_devices() tests ────────────────────────────────────────── + + +def test_bt_midi_connected_devices(mock_subprocess): + """bt_midi_connected_devices returns connected devices.""" + mock_subprocess.side_effect = [ + MagicMock( # paired-devices + stdout="Device AA:BB:CC:DD:EE:FF MIDI Keys\n" + "Device 11:22:33:44:55:66 Pedal\n", + returncode=0, + ), + MagicMock(stdout="", returncode=0), # info - none connected + ] + devices = bt_midi_connected_devices() + assert devices == [] # none connected + + +def test_bt_midi_connected_devices_with_connected(mock_subprocess): + """bt_midi_connected_devices returns only connected devices.""" + with patch("src.system.bluetooth.bt_paired_devices", return_value=[ + {"address": "AA:BB:CC:DD:EE:FF", "name": "MIDI Keys", "connected": True}, + {"address": "11:22:33:44:55:66", "name": "Pedal", "connected": False}, + ]): + devices = bt_midi_connected_devices() + assert len(devices) == 1 + assert devices[0]["name"] == "MIDI Keys" \ No newline at end of file diff --git a/tests/test_network.py b/tests/test_network.py new file mode 100644 index 0000000..78c2ce7 --- /dev/null +++ b/tests/test_network.py @@ -0,0 +1,323 @@ +"""Tests for the WiFi network management module (src.system.network). + +Uses mocking to avoid requiring actual WiFi hardware. + +Run with: + pytest tests/test_network.py -v +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +# Ensure src is on sys.path +SRC = Path(__file__).resolve().parent.parent / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from src.system.network import ( + wifi_scan, + wifi_status, + wifi_saved_networks, + wifi_connect, + wifi_disconnect, + wifi_forget, + hotspot_status, + hotspot_enable, + hotspot_disable, + _has_nmcli, +) + + +# ── Fixtures ───────────────────────────────────────────────────────────────── + + +@pytest.fixture +def mock_subprocess(): + """Patch subprocess.run globally for all tests in this module.""" + with patch("src.system.network.subprocess.run") as mock: + yield mock + + +@pytest.fixture +def mock_hostapd_conf(tmp_path): + """Create a mock /etc/hostapd/hostapd.conf for hotspot tests.""" + conf = tmp_path / "hostapd.conf" + conf.write_text("ssid=Pi-Pedal\nchannel=6\n") + return conf + + +# ── _has_nmcli() tests ──────────────────────────────────────────────────────── + + +def test_has_nmcli_true(mock_subprocess): + """When 'which nmcli' succeeds, _has_nmcli returns True.""" + mock_subprocess.return_value = MagicMock(returncode=0) + assert _has_nmcli() is True + + +def test_has_nmcli_false(mock_subprocess): + """When 'which nmcli' fails, _has_nmcli returns False.""" + mock_subprocess.side_effect = FileNotFoundError("no nmcli") + assert _has_nmcli() is False + + +# ── wifi_scan() tests ───────────────────────────────────────────────────────── + + +def test_wifi_scan_nmcli_networks(mock_subprocess): + """wifi_scan returns parsed networks from nmcli.""" + mock_subprocess.return_value = MagicMock( + stdout="Net1:75:WPA2\nNet2:50:WPA2\nNet3:90:Open\n", + returncode=0, + ) + networks = wifi_scan() + assert len(networks) == 3 + assert networks[0]["ssid"] == "Net3" # sorted by signal desc + assert networks[0]["signal"] == 90 + assert networks[0]["encryption"] == "Open" + + +def test_wifi_scan_nmcli_empty(mock_subprocess): + """wifi_scan returns empty list when no networks found.""" + mock_subprocess.return_value = MagicMock(stdout="", returncode=0) + networks = wifi_scan() + assert networks == [] + + +def test_wifi_scan_nmcli_fallback_iwlist(mock_subprocess): + """wifi_scan falls back to iwlist when nmcli fails.""" + # First call (nmcli) fails + mock_subprocess.side_effect = [ + FileNotFoundError("no nmcli"), # _has_nmcli check + MagicMock(stdout="", returncode=0), # First nmcli scan attempt + ] + # Actually, let's patch _has_nmcli to make this clearer + with patch("src.system.network._has_nmcli", return_value=False): + with patch("src.system.network._iwlist_scan", return_value=[ + {"ssid": "TestNet", "signal": 60, "encryption": "WPA2"} + ]) as mock_iw: + networks = wifi_scan() + assert len(networks) == 1 + assert networks[0]["ssid"] == "TestNet" + assert networks[0]["signal"] == 60 + mock_iw.assert_called_once() + + +def test_wifi_scan_handles_exception(mock_subprocess): + """wifi_scan returns empty list on unexpected error.""" + with patch("src.system.network._has_nmcli", return_value=True): + mock_subprocess.side_effect = RuntimeError("Unexpected") + networks = wifi_scan() + assert networks == [] + + +# ── wifi_status() tests ─────────────────────────────────────────────────────── + + +def test_wifi_status_connected(mock_subprocess): + """wifi_status returns connected state with SSID and IP.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="wlan0:wifi:connected:HomeNet\neth0:ethernet:connected:eth0\n", returncode=0), + MagicMock(stdout="GENERAL.CONNECTION:HomeNet\nIP4.ADDRESS:192.168.1.100/24\n", returncode=0), + MagicMock(stdout="HomeNet:85\nOtherNet:30\n", returncode=0), + ] + with patch("src.system.network._has_nmcli", return_value=True): + status = wifi_status() + assert status["connected"] is True + assert status["ssid"] == "HomeNet" + assert status["ip"] == "192.168.1.100" + assert status["signal"] == 85 + assert status["device"] == "wlan0" + + +def test_wifi_status_not_connected(mock_subprocess): + """wifi_status returns disconnected when no WiFi connection.""" + mock_subprocess.side_effect = [ + MagicMock(stdout="wlan0:wifi:disconnected:\neth0:ethernet:connected:eth0\n", returncode=0), + ] + status = wifi_status() + assert status["connected"] is False + assert status["ssid"] is None + + +def test_wifi_status_no_wifi_device(mock_subprocess): + """wifi_status returns disconnected when no WiFi device exists.""" + mock_subprocess.return_value = MagicMock( + stdout="eth0:ethernet:connected:eth0\n", + returncode=0, + ) + status = wifi_status() + assert status["connected"] is False + + +# ── wifi_connect() tests ────────────────────────────────────────────────────── + + +def test_wifi_connect_requires_params(): + """wifi_connect returns error when SSID or password missing.""" + result = wifi_connect("", "password") + assert result["ok"] is False + assert "SSID" in result["error"] + + result = wifi_connect("MyNet", "") + assert result["ok"] is False + assert "Password" in result["error"] + + +def test_wifi_connect_new_network(mock_subprocess): + """wifi_connect calls nmcli to connect to a new network.""" + # Saved networks (nmcli connection show) returns empty for new + # Has nmcli → True from first call check + with patch("src.system.network._has_nmcli", return_value=True): + mock_subprocess.side_effect = [ + # _nmcli_saved - empty list + MagicMock(stdout="", returncode=0), + # _nmcli_connect - success + MagicMock(stdout="", returncode=0), + ] + result = wifi_connect("NewNet", "secret123") + assert result["ok"] is True + assert mock_subprocess.call_count >= 2 + + +def test_wifi_connect_existing_network(mock_subprocess): + """wifi_connect activates an already-saved network.""" + with patch("src.system.network._has_nmcli", return_value=True): + mock_subprocess.side_effect = [ + # _nmcli_saved - existing network + MagicMock(stdout="NewNet:abc-123:wireless\nOldNet:def-456:wireless\n", returncode=0), + # _nmcli_connect - connection up + MagicMock(stdout="", returncode=0), + ] + result = wifi_connect("NewNet", "secret123") + assert result["ok"] is True + + +# ── wifi_disconnect() tests ─────────────────────────────────────────────────── + + +def test_wifi_disconnect(mock_subprocess): + """wifi_disconnect returns ok.""" + with patch("src.system.network._has_nmcli", return_value=True): + mock_subprocess.side_effect = [ + # _nmcli_status + MagicMock(stdout="wlan0:wifi:connected:HomeNet\n", returncode=0), + MagicMock(stdout="GENERAL.CONNECTION:HomeNet\nIP4.ADDRESS:192.168.1.100/24\n", returncode=0), + MagicMock(stdout="HomeNet:80\n", returncode=0), + # _nmcli_disconnect + MagicMock(stdout="", returncode=0), + ] + result = wifi_disconnect() + assert result["ok"] is True + + +# ── wifi_forget() tests ────────────────────────────────────────────────────── + + +def test_wifi_forget_network(mock_subprocess): + """wifi_forget deletes a saved network.""" + with patch("src.system.network._has_nmcli", return_value=True): + mock_subprocess.side_effect = [ + # _nmcli_saved + MagicMock(stdout="MyNet:abc-123:wireless\nOther:def-456:wireless\n", returncode=0), + # _nmcli_forget - nmcli connection delete + MagicMock(stdout="", returncode=0), + ] + result = wifi_forget("MyNet") + assert result["ok"] is True + + +def test_wifi_forget_not_found(mock_subprocess): + """wifi_forget returns ok=False when network not saved.""" + with patch("src.system.network._has_nmcli", return_value=True): + mock_subprocess.return_value = MagicMock( + stdout="Other:def-456:wireless\n", + returncode=0, + ) + result = wifi_forget("NonExistent") + assert result["ok"] is False + + +# ── wifi_saved_networks() tests ──────────────────────────────────────────────── + + +def test_wifi_saved_networks(mock_subprocess): + """wifi_saved_networks returns saved WiFi connections.""" + with patch("src.system.network._has_nmcli", return_value=True): + mock_subprocess.return_value = MagicMock( + stdout="HomeNet:abc-def:wireless\nOfficeNet:ghi-jkl:wireless\neth0:xyz:ethernet\n", + returncode=0, + ) + saved = wifi_saved_networks() + assert len(saved) == 2 + assert saved[0]["ssid"] == "HomeNet" + assert saved[1]["ssid"] == "OfficeNet" + + +# ── hotspot_status() tests ──────────────────────────────────────────────────── + + +def test_hotspot_status_active(mock_subprocess, mock_hostapd_conf): + """hotspot_status returns active state when hostapd is running.""" + with patch("src.system.network.Path.exists", return_value=True): + with patch("src.system.network.Path.read_text", return_value="ssid=Pi-Pedal\nchannel=6\n"): + mock_subprocess.side_effect = [ + MagicMock(stdout="active\n", returncode=0), # systemctl is-active + MagicMock(stdout="Station aa:bb:cc:dd:ee:ff\n", returncode=0), # iw station dump + ] + status = hotspot_status() + assert status["active"] is True + assert status["ssid"] == "Pi-Pedal" + assert status["clients"] == 1 + assert status["ip"] == "192.168.4.1" + + +def test_hotspot_status_inactive(mock_subprocess): + """hotspot_status returns inactive when hostapd is stopped.""" + mock_subprocess.return_value = MagicMock(stdout="inactive\n", returncode=0) + status = hotspot_status() + assert status["active"] is False + assert status["clients"] == 0 + + +# ── hotspot_enable() / hotspot_disable() tests ───────────────────────────────── + + +def test_hotspot_enable_short_password(): + """hotspot_enable rejects passwords shorter than 8 chars.""" + result = hotspot_enable("TestAP", "123") + assert result["ok"] is False + assert "at least 8 characters" in result["error"].lower() + + +def test_hotspot_enable_success(mock_subprocess): + """hotspot_enable calls the setup script.""" + mock_subprocess.return_value = MagicMock(returncode=0) + result = hotspot_enable("TestAP", "longpassword") + assert result["ok"] is True + + +def test_hotspot_enable_failure(mock_subprocess): + """hotspot_enable returns error on script failure.""" + mock_subprocess.return_value = MagicMock(returncode=1, stderr="Script failed") + result = hotspot_enable("TestAP", "longpassword") + assert result["ok"] is False + + +def test_hotspot_disable_success(mock_subprocess): + """hotspot_disable calls the disable script.""" + mock_subprocess.return_value = MagicMock(returncode=0) + result = hotspot_disable() + assert result["ok"] is True + + +def test_hotspot_disable_failure(mock_subprocess): + """hotspot_disable returns error on script failure.""" + mock_subprocess.return_value = MagicMock(returncode=1, stderr="Script failed") + result = hotspot_disable() + assert result["ok"] is False \ No newline at end of file