Bluetooth Scanner Dashboard

# pip install bleak websockets
import asyncio
import json
from bleak import BleakScanner
import websockets

connected_clients = set()

async def scan_callback(device, advertisement_data):
    device_data = {
        "address": device.address,
        "name": device.name or advertisement_data.local_name or "Unbekannt",
        "rssi": device.rssi,
        "manufacturer_data": {str(k): v.hex() for k, v in advertisement_data.manufacturer_data.items()},
        "service_uuids": advertisement_data.service_uuids,
        "tx_power": advertisement_data.tx_power
    }
    if connected_clients:
        message = json.dumps({"type": "device", "data": device_data})
        await asyncio.gather(*[client.send(message) for client in connected_clients], return_exceptions=True)

async def handler(websocket):
    connected_clients.add(websocket)
    print(f"Client verbunden: {websocket.remote_address}")
    try:
        async for message in websocket:
            data = json.loads(message)
            if data.get("action") == "start":
                print("Starte kontinuierlichen BLE-Scan...")
            elif data.get("action") == "stop":
                print("Stoppe BLE-Scan...")
    except websockets.exceptions.ConnectionClosed:
        pass
    finally:
        connected_clients.remove(websocket)
        print("Client getrennt")

async def main():
    scanner = BleakScanner(detection_callback=scan_callback)
    await scanner.start()
    async with websockets.serve(handler, "localhost", 8765):
        print("WebSocket Server gestartet auf ws://localhost:8765")
        await asyncio.Future()  # läuft endlos

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

Kopieren