如何实现RTL-SDR异步流在N秒后终止并让程序正常退出？

我太懂你这种折腾的痛苦了——试了三种方法，要么程序挂着不退出，要么报错，要么根本没正确读取样本。咱们先拆解下你每个尝试的问题，再给你两个靠谱的解决方案。

先说说你之前尝试的问题出在哪 #

• Attempt 1：你用asyncio.wait等第一个任务完成，但streaming任务还在后台等着下一批样本呢！虽然你调用了sdr.stop()，但异步迭代器没被取消，程序就卡在那等它结束，自然不会退出。
• Attempt 2：你在async for循环里直接调用sdr.stop()，这时候迭代器还在拼命从设备读数据，设备突然被停了，LibUSB当然会报“找不到实体”的错——就像你正在喝水，别人突然把杯子拿走了，你肯定也懵。
• Attempt 3：你直接打印sdr.stream()得到的只是迭代器对象，根本没在读取样本！而且aclose()也不是正确的停止方式，完全没用到异步流的正确逻辑。

解决方案1：用asyncio.wait_for直接给流任务加超时（最简洁） #

这个方法利用asyncio.wait_for自带的超时机制，超时后自动取消流任务，我们只需要捕获异常并正确关闭设备就行：

import asyncio
from rtlsdr import RtlSdr

async def streaming(sdr):
    async for samples in sdr.stream():
        print(samples)

async def main():
    sdr = RtlSdr()
    try:
        # 给流任务设置5秒超时
        await asyncio.wait_for(streaming(sdr), timeout=5)
    except asyncio.TimeoutError:
        # 超时后优雅停止设备
        await sdr.stop()
        sdr.close()
        print("5秒超时，流已终止，程序正常退出")

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

解决方案2：用asyncio.Event手动触发终止（更灵活） #

如果以后你需要根据其他条件终止流（比如收到某个信号、检测到特定样本），用事件触发会更灵活：

import asyncio
from rtlsdr import RtlSdr

async def streaming(sdr, stop_event):
    async for samples in sdr.stream():
        # 每次迭代检查是否需要停止
        if stop_event.is_set():
            break
        print(samples)
    # 跳出循环后再停止设备
    await sdr.stop()
    sdr.close()

async def timeout_trigger(stop_event, duration):
    await asyncio.sleep(duration)
    stop_event.set()
    print("时间到，触发流终止")

async def main():
    sdr = RtlSdr()
    stop_event = asyncio.Event()
    
    # 创建流任务和超时触发任务
    stream_task = asyncio.create_task(streaming(sdr, stop_event))
    timeout_task = asyncio.create_task(timeout_trigger(stop_event, 5))
    
    # 等待两个任务都完成
    await asyncio.gather(stream_task, timeout_task)

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

如果你想拯救Attempt 1 #

其实给Attempt 1加个任务取消的逻辑也能正常工作，就是稍显繁琐：

import asyncio
import datetime
from rtlsdr import RtlSdr

starttime = datetime.datetime.now()
stoptime = starttime + datetime.timedelta(0,5)

async def timeout():
    while datetime.datetime.now() < stoptime:
        await asyncio.sleep(0.5)

async def streaming(sdr):
    async for samples in sdr.stream():
        print(samples)

async def main():
    sdr = RtlSdr()
    task1 = asyncio.create_task(timeout())
    task2 = asyncio.create_task(streaming(sdr))
    
    done, pending = await asyncio.wait({task1, task2}, return_when=asyncio.FIRST_COMPLETED)
    
    # 取消未完成的任务
    for task in pending:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
    
    sdr.stop()
    sdr.close()

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

这样修改后，timeout任务完成后，会取消还在跑的streaming任务，程序就能正常退出了。

备注：内容来源于stack exchange，提问作者AaronJPung