SDR/model/jammer_radar.py

from typing import Tuple
import numpy as np
from scipy.signal import find_peaks

class JammerRadar:
    def __init__(self, usrp, rx, tx):
        self.usrp = usrp
        self.rx = rx
        self.tx = tx

    SAMPLING_RATE = 10e6
    CENTER_FREQ = 6e9
    GAIN = 50

    def generate_jamming_signal(self, algorithm: callable, bandwidth: float, duration: float, sample_rate=SAMPLING_RATE, gain=GAIN) -> Tuple[np.ndarray, float, float]:
        # 应用干扰算法
        signal, lower_bound, upper_bound, *_ = algorithm(  # 使用*_接收多余返回值
            bandwidth=bandwidth,
            duration=duration,
            sample_rate=sample_rate
        )
        return signal, lower_bound, upper_bound

    def execute_jamming(self, algorithm: callable, bandwidth: float, sample_rate: float, duration: float, gain: float) -> Tuple[np.ndarray, float, float]:
        """
        执行自主干扰生成
        :param algorithm: 干扰算法
        :param bandwidth: 信号带宽 (Hz)
        :param sample_rate: 采样率 (Hz)
        :param duration: 信号持续时间 (秒)
        :param gain: 增益 (dB)
        """
        return self.generate_jamming_signal(
            algorithm=algorithm,
            bandwidth=bandwidth,
            sample_rate=sample_rate,
            duration=duration,
            gain=gain
        )

    #封装一个发送信号的函数
    def send_signal(self, tx_signal, duration, center_freq=CENTER_FREQ, sample_rate=SAMPLING_RATE, gain=GAIN):
        # 发送信号
        self.usrp.send_waveform(tx_signal, duration, center_freq, sample_rate, [self.tx], gain)
        print('干扰雷达已发送信号')

    # 封装一个接收信号的函数
    def recv_signal(self, num_samples, sample_rate=SAMPLING_RATE, center_freq=CENTER_FREQ):
        rx_signal = self.usrp.recv_num_samps(num_samples, center_freq, sample_rate, [self.rx])
        print('干扰雷达已接收信号')
        return rx_signal

    def send_jamming_signal(self, algorithm, bandwidth, duration, sample_rate=SAMPLING_RATE, gain=GAIN):
        jamming_signal, min_val, max_val = self.generate_jamming_signal(
            algorithm=algorithm,
            bandwidth=bandwidth,
            sample_rate=sample_rate,
            duration=duration,
            gain=gain
        )
        self.usrp.send_waveform(jamming_signal, duration, bandwidth, sample_rate, [self.tx], gain)
        return min_val, max_val