本文介绍如何在已存在 main()（非 async）的 Python 项目中，集成异步定时频率切换逻辑——通过 asyncio.create_task 启动后台周期性任务，并在主协程结束时优雅取消，避免阻塞主线程且兼容现有 SDR 数据流架构。
本文介绍如何在已存在 `main()`（非 async）的 python 项目中，集成异步定时频率切换逻辑——通过 `asyncio.create_task` 启动后台周期性任务，并在主协程结束时优雅取消，避免阻塞主线程且兼容现有 sdr 数据流架构。
要在基于 asyncio 的 SDR 数据采集流程中实现 freq1 与 freq2 的周期性自动交换（例如每 5 秒切换一次），关键不在于“在同步函数里 await”，而是在已启动的异步事件循环中并发运行一个长期存活的定时任务。由于你的 main() 是同步入口，但最终调用了 asyncio.run(pipeline(...))，因此所有异步逻辑必须在 pipeline 及其子协程中组织。
以下为推荐实现方案：
✅ 步骤一：定义频率交换协程（后台守护任务）
async def frequency_swapper(config: SampleConfig) -> None:
"""持续运行的协程：按指定间隔交换 freq1 和 freq2"""
while True:
await asyncio.sleep(config.freq_change_interval)
config.freq1, config.freq2 = config.freq2, config.freq1
print(f"[INFO] Frequency swapped: freq1={config.freq1:.3f} MHz, freq2={config.freq2:.3f} MHz")
⚠️ 注意：该协程必须被
create_task显式调度，否则await frequency_swapper(...)会阻塞当前协程，违背“后台运行”初衷。
✅ 步骤二：在 pipeline() 中动态启停任务
修改你的 pipeline 函数（需声明为 async def），加入任务生命周期管理：
async def pipeline(
process_config,
source_gen,
task_results
):
# 启动频率交换任务（仅当启用时）
freq_task = None
if hasattr(process_config, "freq_change_interval") and process_config.freq_change_interval > 0:
freq_task = asyncio.create_task(frequency_swapper(process_config))
try:
# ? 这里是你原有的数据处理逻辑（如 source_gen 迭代、task_results 调用等）
async for sample_batch in source_gen:
result = await task_results(sample_batch, process_config)
# ... 处理 result
finally:
# 确保无论正常结束或异常退出，都清理后台任务
if freq_task and not freq_task.done():
freq_task.cancel()
try:
await freq_task # 等待取消完成（可选，提升健壮性）
except asyncio.CancelledError:
pass
✅ 步骤三：确保 SampleConfig 支持就地修改
确认 SampleConfig 是可变对象（如普通 dataclass 或 namedtuple 的替代类）。若当前是 namedtuple 或 frozen=True dataclass，需改为可变结构：
from dataclasses import dataclass
@dataclass
class SampleConfig:
freq1: float
freq2: float
freq_change_interval: float = 0.0
# 其他字段...
? 补充说明与最佳实践
-
不要在
main()中调用asyncio.run()多次：你已在airspy分支中正确使用asyncio.run(pipeline(...))—— 这是启动整个异步流程的唯一入口，所有子任务（包括frequency_swapper）都应在此事件循环内并发执行。 -
避免竞态条件：
freq1/freq2被多个协程（如source_radio读取器、task_results处理器）共享时，建议添加asyncio.Lock（若访问频繁且敏感），但对 SDR 配置参数通常只需保证写入原子性（tuple赋值本身是线程/协程安全的）。 -
调试技巧：在
frequency_swapper中加入日志，确认任务是否如期触发；使用asyncio.all_tasks()可检查当前活跃任务数。
通过以上设计，你无需改造 main() 的同步本质，即可在保持代码清晰性的同时，精准实现“后台定时参数切换”这一典型异步模式。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
