WellNuo/ble-wifi-setup.py

#!/usr/bin/env python3
"""
BLE WiFi Setup for WellNuo WP sensors
Single characteristic for read/write/notify
"""

import asyncio
import sys
from bleak import BleakClient, BleakScanner

# Sensor BLE UUIDs
SERVICE_UUID = "4fafc201-1fb5-459e-8fcc-c5c9c331914b"
CHAR_UUID = "beb5483e-36e1-4688-b7f5-ea07361b26a8"  # Single characteristic for all operations

DEVICE_PIN = "7856"

# Global for notification response
response_data = None
response_event = asyncio.Event()

def notification_handler(sender, data):
    """Handle notifications from sensor"""
    global response_data
    decoded = data.decode('utf-8', errors='replace')
    print(f"  [RX] {decoded}")
    response_data = decoded
    response_event.set()

async def send_command(client, command, timeout=15):
    """Send command and wait for response via notification"""
    global response_data
    response_data = None
    response_event.clear()

    print(f"  [TX] {command}")
    await client.write_gatt_char(CHAR_UUID, command.encode('utf-8'))

    try:
        await asyncio.wait_for(response_event.wait(), timeout=timeout)
        return response_data
    except asyncio.TimeoutError:
        print(f"  [TIMEOUT] No response after {timeout}s")
        # Try reading directly
        try:
            data = await client.read_gatt_char(CHAR_UUID)
            decoded = data.decode('utf-8', errors='replace')
            print(f"  [READ] {decoded}")
            return decoded
        except:
            return None

async def get_wifi_list(client):
    """Get list of available WiFi networks"""
    print("\n=== Scanning WiFi networks ===")
    response = await send_command(client, "W|list", timeout=20)
    if response and "|W|list|" in response:
        parts = response.split("|W|list|")
        if len(parts) > 1:
            networks_str = parts[1]
            networks = networks_str.split("|")
            print("\nAvailable WiFi networks:")
            for i, net in enumerate(networks, 1):
                if "," in net:
                    ssid, rssi = net.rsplit(",", 1)
                    try:
                        rssi_val = int(rssi)
                        signal = "Strong" if rssi_val > -50 else "Good" if rssi_val > -70 else "Weak"
                    except:
                        signal = ""
                    print(f"  {i}. {ssid} (RSSI: {rssi} {signal})")
            return networks
    return []

async def setup_wifi(client, ssid, password):
    """Configure WiFi on sensor"""
    print(f"\n=== Configuring WiFi ===")
    print(f"SSID: {ssid}")
    print(f"Password: {'*' * len(password)} ({len(password)} chars)")

    # Step 1: Unlock with PIN
    print("\n1. Unlocking sensor...")
    response = await send_command(client, f"pin|{DEVICE_PIN}")
    if not response or "ok" not in response.lower():
        print(f"   ERROR: Unlock failed! Response: {response}")
        return False
    print("   Unlocked!")

    # Small delay after unlock
    await asyncio.sleep(0.5)

    # Step 2: Send WiFi credentials
    print(f"\n2. Sending WiFi credentials...")
    cmd = f"W|{ssid},{password}"
    response = await send_command(client, cmd, timeout=30)

    if response:
        if "|W|ok" in response.lower() or "wifi|ok" in response.lower() or response.endswith("|ok"):
            print("   SUCCESS! WiFi configured.")
            return True
        elif "|W|fail" in response.lower() or "fail" in response.lower():
            print(f"   FAILED! Sensor rejected credentials.")
            print(f"   Response: {response}")
            return False

    print("   Unknown response or timeout")
    return False

async def read_status(client):
    """Read current sensor status"""
    print("\n=== Current sensor status ===")
    try:
        data = await client.read_gatt_char(CHAR_UUID)
        decoded = data.decode('utf-8', errors='replace')
        print(f"Status: {decoded}")
        return decoded
    except Exception as e:
        print(f"Error reading status: {e}")
        return None

async def scan_sensors():
    """Scan for WP sensors"""
    print("Scanning for BLE devices...")
    devices = await BleakScanner.discover(timeout=5.0)

    wp_devices = []
    for d in devices:
        if d.name and d.name.startswith("WP_"):
            wp_devices.append(d)
            print(f"  Found: {d.name} ({d.address})")

    return wp_devices

async def main():
    print("=" * 60)
    print("WellNuo Sensor WiFi Setup Tool")
    print("=" * 60)

    # Scan for sensors
    devices = await scan_sensors()
    if not devices:
        print("\nNo WP sensors found. Make sure sensor is powered on and in range.")
        return

    device = devices[0]
    print(f"\nConnecting to {device.name}...")

    async with BleakClient(device.address) as client:
        print("Connected!")

        # Start notifications
        await client.start_notify(CHAR_UUID, notification_handler)

        # Read current status
        await read_status(client)

        if len(sys.argv) < 2:
            # Just list networks
            await get_wifi_list(client)

            print("\n" + "=" * 60)
            print("Usage:")
            print("  python ble-wifi-setup.py list              # List WiFi networks")
            print("  python ble-wifi-setup.py <SSID> <PASSWORD> # Configure WiFi")
            print("=" * 60)

        elif sys.argv[1] == "list":
            await get_wifi_list(client)

        elif len(sys.argv) >= 3:
            ssid = sys.argv[1]
            password = sys.argv[2]

            # First check if network is visible
            print("\nChecking if network is visible...")
            networks = await get_wifi_list(client)

            network_found = False
            for net in networks:
                if "," in net:
                    net_ssid = net.rsplit(",", 1)[0]
                    if net_ssid.lower() == ssid.lower():
                        network_found = True
                        if net_ssid != ssid:
                            print(f"\nNote: Exact SSID is '{net_ssid}' (case matters!)")
                            ssid = net_ssid
                        break

            if not network_found:
                print(f"\nWARNING: Network '{ssid}' not found in scan!")
                print("The sensor might not see this network.")
                print("Possible reasons:")
                print("  - Network is 5GHz only (sensor needs 2.4GHz)")
                print("  - Network is hidden")
                print("  - Sensor too far from router")

            # Try to configure anyway
            success = await setup_wifi(client, ssid, password)

            if success:
                print("\n" + "=" * 60)
                print("WiFi setup SUCCESSFUL!")
                print("Sensor should now connect to the network.")
                print("=" * 60)
            else:
                print("\n" + "=" * 60)
                print("WiFi setup FAILED!")
                print("Possible issues:")
                print("  1. Wrong password (case-sensitive!)")
                print("  2. Network is 5GHz only")
                print("  3. Sensor can't reach the router")
                print("=" * 60)

        await client.stop_notify(CHAR_UUID)

if __name__ == "__main__":
    asyncio.run(main())