GNURadio自定义信号源模块FFT自动切换问题求助

Hey there, let's break down and fix this GNURadio module issue you're dealing with!

The Core Problem #

Your current code tries to loop through all 7 hop frequencies in a single work() call, but GNURadio's work() function operates on batches of samples. By the time you hit the return statement, you've overwritten the entire output_items array with the last frequency's waveform—so the FFT only sees that final frequency. The loop runs entirely within one work() execution, which doesn't give the flowgraph time to update the FFT display between hops.

The Fix: Persist Hop State & Process Samples in Chunks #

Instead of trying to cram all hops into one work() call, we need to track the current hop state across multiple work() invocations. Here's how to adjust your code:

1. Add State Variables to Your Module

First, initialize persistent state in your module's __init__ method. These variables will remember where you left off between work() calls:

def __init__(self, ...):
    # Your existing initialization code here
    self.current_hop_idx = 0  # Tracks which hop we're on
    self.remaining_hop_samples = 0  # Samples left to send for current hop
    self.current_payload_step = 0  # Step value for current frequency
    self.index_f = 0  # Persistent phase index (move from work() here)
    self.hop_duration_samples = int(4096 * 0.1)  # 0.1 sec per hop (adjust as needed)
    
    # Pre-generate SINTAB once (no need to re-make it every work call!)
    self.SINTAB = np.zeros((4096), dtype=complex)
    for i in range(4096):
        self.SINTAB[i] = math.sin(2 * math.pi * i / 4096) - 1j * math.cos(2 * math.pi * i / 4096)

2. Rewrite the work() Function

Now modify work() to process samples in chunks, switching hops only when the current hop's samples are exhausted:

def work(self, input_items, output_items):
    out = output_items[0]
    total_samples_to_write = len(out)
    samples_written = 0

    while samples_written < total_samples_to_write:
        # If we've finished the current hop, switch to the next one
        if self.remaining_hop_samples == 0:
            # Loop back to start of hop list if we've gone through all 7
            if self.current_hop_idx >= len(sub_seed_hop_freq):
                self.current_hop_idx = 0
            
            # Get current hop's frequency
            check = sub_seed_hop_freq[self.current_hop_idx]
            if check == 1:
                Freq = 0.5e6
            elif check == 2:
                Freq = 1e6
            elif check == 3:
                Freq = 1.5e6
            else:
                Freq = 0  # Default fallback frequency
            
            # Calculate payload_step using your existing conversion logic
            self.current_payload_step = (Freq / 4096) * 4096  # Replace with your actual formula
            
            # Set how many samples to send for this hop
            self.remaining_hop_samples = self.hop_duration_samples
            self.current_hop_idx += 1
        
        # Calculate how many samples we can write in this iteration
        samples_to_write_now = min(self.remaining_hop_samples, total_samples_to_write - samples_written)
        
        # Fill the output buffer with the current hop's waveform
        for i in range(samples_to_write_now):
            out[samples_written + i] = self.SINTAB[int(round(self.index_f))]
            self.index_f += self.current_payload_step
            # Wrap phase index to stay within SINTAB bounds
            if self.index_f >= len(self.SINTAB):
                self.index_f -= len(self.SINTAB)
        
        # Update counters
        samples_written += samples_to_write_now
        self.remaining_hop_samples -= samples_to_write_now

    return total_samples_to_write

Key Changes Explained #

• Persistent State: Variables like self.current_hop_idx and self.remaining_hop_samples keep track of your progress between work() calls, so each hop gets its turn to send samples over multiple flowgraph iterations.
• Chunked Processing: Instead of overwriting the entire output buffer with one frequency, we fill it in pieces—each piece corresponding to a single hop. This gives the FFT time to update and display each hop as it's sent.
• Efficient SINTAB Generation: We pre-make the SINTAB once in __init__ instead of regenerating it every time work() runs, which saves computation time.

Quick Adjustments to Tweak Behavior #

• Change self.hop_duration_samples to adjust how long each hop lasts (e.g., 4096 * 0.001 for 1ms per hop).
• Double-check your payload_step calculation to ensure it correctly maps frequency to the phase step needed for your SINTAB.

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