在非异步main函数的Python SDR项目中实现异步定时器驱动的频率参数切换

首先，你的main函数虽然是同步的，但核心的采样逻辑已经通过asyncio.run跑在异步事件循环里了，所以我们只需要把频率切换的逻辑也放到这个异步环境里，和主采样任务并行执行就好，不用改动main的同步结构。下面是贴合你现有代码的具体修改步骤：

1. 调整SampleConfig，支持异步安全的频率修改 #

首先要让SampleConfig的频率参数可以被动态修改，并且保证异步环境下的读写安全（避免采样逻辑和切换逻辑同时修改/读取频率导致的竞态问题）。我们给它加一个异步锁：

如果你的SampleConfig是dataclass：

from dataclasses import dataclass, field
import asyncio

@dataclass(frozen=False)  # frozen=False允许修改属性
class SampleConfig:
    # 原来的类级默认值
    freq1: float = ...  # 替换成你实际的默认值
    freq2: float = ...
    freq_change_interval: float = ...
    # 新增异步锁，用于频率读写的同步（不需要在init时传入）
    freq_lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)

如果是普通类：

import asyncio

class SampleConfig:
    # 原来的类级默认值
    freq1 = ...
    freq2 = ...
    freq_change_interval = ...

    def __init__(self, freq1, freq2, freq_change_interval):
        self.freq1 = freq1
        self.freq2 = freq2
        self.freq_change_interval = freq_change_interval
        self.freq_lock = asyncio.Lock()  # 初始化异步锁

2. 编写异步的频率切换任务 #

这个任务会循环等待指定的间隔，然后安全地交换两个频率：

async def frequency_switcher_task(sample_config):
    while True:
        # 等待指定的切换间隔
        await asyncio.sleep(sample_config.freq_change_interval)
        # 用异步锁保护频率交换操作，避免和采样逻辑的读写冲突
        async with sample_config.freq_lock:
            sample_config.freq1, sample_config.freq2 = sample_config.freq2, sample_config.freq1
            print(f"已切换频率: freq1={sample_config.freq1:.2f}, freq2={sample_config.freq2:.2f}")

3. 修改主异步入口，并行运行采样和切换任务 #

原来的代码是直接用asyncio.run启动pipeline，现在我们要把切换任务也加入事件循环，和主采样任务一起运行：

首先在模块里新增一个异步入口函数：

async def run_sampling_and_switcher(sample_config, process_config, source_gen, task_results):
    switcher_task = None
    # 只有当间隔大于0时才启动切换任务
    if sample_config.freq_change_interval > 0:
        switcher_task = asyncio.create_task(frequency_switcher_task(sample_config))
    
    try:
        # 运行主采样pipeline
        await pipeline(
            process_config=process_config,
            source_gen=source_gen,
            task_results=task_results,
        )
    finally:
        # 程序退出时优雅地取消切换任务
        if switcher_task:
            switcher_task.cancel()
            try:
                await switcher_task
            except asyncio.CancelledError:
                print("频率切换任务已停止")

然后在main函数的airspy分支，替换原来的asyncio.run调用：

# 替换原来的asyncio.run(pipeline(...))部分
asyncio.run(
    run_sampling_and_switcher(
        sample_config=sample_config,
        process_config=process_config,
        source_gen=source_radio(
            reader=SampleReader(sample_config),  # 确保reader持有SampleConfig的引用，而非缓存初始频率
            buffer=SampleBuffer(),
            num_samples_to_process=process_config.num_samples_to_process,
        ),
        task_results=results_pipeline,
    )
)

4. 确保采样Reader使用最新的频率值 #

关键一步：你的SampleReaderAirspy（或其他radio的reader）不能在初始化时缓存频率，而要在每次采样前从SampleConfig获取最新值：

class SampleReaderAirspy:
    def __init__(self, sample_config):
        self.sample_config = sample_config  # 保存SampleConfig引用
        # 其他初始化逻辑...

    async def read_samples(self):
        # 每次采样前获取最新频率，用锁保证安全
        async with self.sample_config.freq_lock:
            current_freq1 = self.sample_config.freq1
            current_freq2 = self.sample_config.freq2
        
        # 用current_freq1和current_freq2配置SDR进行采样
        # 你的采样逻辑代码...

这样每次采样时，都会拿到最新的、可能刚切换后的频率值。

5. 边界情况处理 #

• 如果freq_change_interval设为0，切换任务不会启动，保持初始频率不变
• 程序退出时，finally块会自动取消切换任务，避免残留异步任务导致的问题

这样修改后，就能完全实现你想要的“每隔指定间隔自动交换freq1和freq2”的逻辑，完美贴合你现有代码的异步+同步混合结构。

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