Flask+Bleak开发蓝牙Web应用遇参数缺失错误求助

问题描述 #

开发Web应用时，计划通过Bleak库获取蓝牙数据，后续打算存入浏览器Cookie/Session，目前仅需在浏览器查看数据，但触发错误：
TypeError: devices() missing 2 required positional arguments: 'device' and 'advertisement_data'

原代码 #

from flask import Flask
import asyncio
from bleak import BleakScanner
from uuid import UUID
import json
from construct import Array, Byte, Const, Int8sl, Int16ub, Struct
from construct.core import ConstError
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData

app = Flask(__name__)

ibeacon_format = Struct(
    "type_length" / Const(b"\x02\x15"),
    "uuid" / Array(16, Byte),
    "major" / Int16ub,
    "minor" / Int16ub,
    "power" / Int8sl,
)

class UUIDEncoder(json.JSONEncoder):
    def default(self, uuid):
        if isinstance(uuid, UUID):
            return uuid.hex
        return json.JSONEncoder.default(self, uuid)

@app.route("/")
def hello_world():
    return "<p>Hello, World!</p>"

@app.route("/beacons")
async def devices(device: BLEDevice, advertisement_data: AdvertisementData):
    try:
        macadress = device.address
        name = advertisement_data.local_name
        apple_data = advertisement_data.manufacturer_data[0x004C]
        ibeacon = ibeacon_format.parse(apple_data)
        uuid = UUID(bytes=bytes(ibeacon.uuid))
        minor = ibeacon.minor
        major = ibeacon.major
        power = ibeacon.power
        rssi = device.rssi
        rssi = int(rssi)

        beacons = {
            "Mac Adress": macadress,
            "Local Name": name,
            "UUID": uuid,
            "Major": major,
            "Minor": minor,
            "TX Power": power,
            "RSSI": rssi
        }

        if (beacons["Local Name"] == "POI" and beacons["RSSI"] <= -40 and beacons["RSSI"] >= -80):
            return print(beacons)
        else:
            pass

    except KeyError:
        pass
    except ConstError:
        pass

async def main():
    scanner = BleakScanner()
    scanner.register_detection_callback(devices)
    while (True):
        await scanner.start()
        await asyncio.sleep(0.5)
        await scanner.stop() 

if __name__ == '__main__':
    app.run(host="127.0.0.0", port=5000, debug=True)

问题根源 #

1. 角色冲突：将Bleak的设备检测回调函数同时标记为Flask路由函数，两者参数要求完全矛盾——Flask路由不需要device和advertisement_data参数，而Bleak回调必须接收这两个参数。
2. 扫描逻辑未执行：Flask启动后，main()异步函数从未被调用，蓝牙扫描完全没启动。
3. 路由响应错误：return print(beacons)无效，print()返回None，Flask路由必须返回合法的HTTP响应内容。

修复方案 #

拆分路由与回调函数，用全局变量存储扫描数据，让Flask路由可以返回数据给浏览器，同时正确启动后台扫描任务：

修改后完整代码 #

from flask import Flask, jsonify
import asyncio
from bleak import BleakScanner
from uuid import UUID
from construct import Array, Byte, Const, Int8sl, Int16ub, Struct
from construct.core import ConstError
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData

app = Flask(__name__)
# 全局变量存储符合条件的beacon数据
detected_beacons = []

ibeacon_format = Struct(
    "type_length" / Const(b"\x02\x15"),
    "uuid" / Array(16, Byte),
    "major" / Int16ub,
    "minor" / Int16ub,
    "power" / Int8sl,
)

@app.route("/")
def hello_world():
    return "<p>Hello, World!</p>"

@app.route("/beacons")
def get_beacons():
    # 返回存储的beacon数据
    return jsonify(detected_beacons)

# 单独的Bleak设备检测回调函数
async def device_detected(device: BLEDevice, advertisement_data: AdvertisementData):
    global detected_beacons
    try:
        mac_address = device.address
        name = advertisement_data.local_name
        apple_data = advertisement_data.manufacturer_data[0x004C]
        ibeacon = ibeacon_format.parse(apple_data)
        uuid = UUID(bytes=bytes(ibeacon.uuid))
        minor = ibeacon.minor
        major = ibeacon.major
        power = ibeacon.power
        rssi = int(device.rssi)

        beacon = {
            "Mac Address": mac_address,
            "Local Name": name,
            "UUID": uuid.hex,  # 直接转成字符串避免JSON序列化问题
            "Major": major,
            "Minor": minor,
            "TX Power": power,
            "RSSI": rssi
        }

        # 过滤符合条件的设备，去重避免重复添加
        if (name == "POI" and -80 <= rssi <= -40):
            exists = any(b["Mac Address"] == mac_address for b in detected_beacons)
            if not exists:
                detected_beacons.append(beacon)
                print(beacon)

    except KeyError:
        # 无苹果厂商数据，跳过
        pass
    except ConstError:
        # 不是iBeacon格式，跳过
        pass

async def scan_task():
    """后台蓝牙扫描任务"""
    scanner = BleakScanner()
    scanner.register_detection_callback(device_detected)
    while True:
        await scanner.start()
        await asyncio.sleep(0.5)
        await scanner.stop()

async def run_app():
    """启动Flask和扫描任务"""
    # 创建扫描任务
    scan = asyncio.create_task(scan_task())
    # 启动Flask（关闭自动重载避免重复启动扫描）
    await app.run(host="127.0.0.1", port=5000, debug=True, use_reloader=False)
    await scan

if __name__ == '__main__':
    asyncio.run(run_app())

关键修改说明 #

• 拆分函数：将原devices拆分为get_beacons（Flask路由，返回数据）和device_detected（Bleak回调，收集数据）。
• 数据存储：用全局列表detected_beacons存储符合条件的设备，路由函数读取该列表返回JSON。
• 正确启动任务：用asyncio.create_task启动后台扫描任务，结合Flask的异步运行方式。
• 修复错误：修正host="127.0.0.0"为127.0.0.1，将UUID直接转为字符串避免序列化问题，添加设备去重逻辑。

内容的提问来源于stack exchange，提问作者profundusman