commit

.idea/SDR.iml

@@ -5,7 +5,7 @@
   </component>
   <component name="NewModuleRootManager">
     <content url="file://$MODULE_DIR$" />
-    <orderEntry type="inheritedJdk" />
+    <orderEntry type="jdk" jdkName="Python 3.9 (SDR)" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
   <component name="TemplatesService">

.idea/encodings.xml

@@ -1,6 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
   <component name="Encoding" defaultCharsetForPropertiesFiles="UTF-8">
+    <file url="file://$PROJECT_DIR$/requirements.txt" charset="UTF-8" />
     <file url="PROJECT" charset="UTF-8" />
   </component>
 </project>

.idea/misc.xml

@@ -3,5 +3,5 @@
   <component name="Black">
     <option name="sdkName" value="SDR" />
   </component>
-  <component name="ProjectRootManager" version="2" project-jdk-name="SDR" project-jdk-type="Python SDK" />
+  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.9 (SDR)" project-jdk-type="Python SDK" />
 </project>

algo/anti_jamming_signal_algo.py

@@ -4,31 +4,6 @@ from typing import Tuple, List
 
 
 class AntiJammingSignalAlgo:
-    @staticmethod
-    def time_frequency_filter(rx_signal, sample_rate):
-        """时频域联合滤波算法"""
-        # 输入信号维度检查
-        if rx_signal.size == 0:
-            return np.zeros_like(rx_signal)
-            
-        # 确保信号为1维数组
-        rx_signal = np.atleast_1d(rx_signal).ravel()
-        
-        spectrum = np.fft.fft(rx_signal)
-        filtered_spectrum = spectrum * np.hamming(len(spectrum))
-        return np.fft.ifft(filtered_spectrum).real
-
-    @staticmethod
-    def adaptive_filter(rx_signal, reference_signal):
-        """自适应滤波算法"""
-        # 实现基于LMS的自适应滤波
-        lms_step = 0.01
-        weights = np.zeros(10)
-        for i in range(len(rx_signal)-10):
-            x = rx_signal[i:i+10]
-            error = reference_signal[i] - np.dot(weights, x)
-            weights += lms_step * error * x
-        return np.convolve(weights, rx_signal, mode='valid')
 
     @staticmethod
     def polarization_filter(rx_signal, sample_rate, polarization_params):
@@ -46,3 +21,57 @@ class AntiJammingSignalAlgo:
         # 应用极化滤波
         polarized_signal = filter_matrix @ rx_signal.reshape(2, -1)
         return polarized_signal[0].real
+
+    @staticmethod
+    def frequency_agility(tx_signal, sample_rate, hop_sequence):
+        """频率捷变算法
+        Args:
+            tx_signal: 原始发射信号
+            sample_rate: 采样率
+            hop_sequence: 跳频序列 (包含频率点和驻留时间)
+        """
+        processed_signal = np.zeros_like(tx_signal)
+        t = 0
+        
+        for freq, dwell in hop_sequence:
+            n_samples = 100
+            freq_shift = np.exp(1j*2*np.pi*freq/sample_rate*np.arange(n_samples))
+            segment = tx_signal[t:t+n_samples] * freq_shift
+            processed_signal[t:t+n_samples] = segment
+            t += n_samples
+            if t >= len(tx_signal):
+                break
+        return processed_signal
+
+    @staticmethod
+    def waveform_agility(tx_signal, sample_rate, waveform_params):
+        """波形捷变算法
+        Args:
+            tx_signal: 原始发射信号
+            sample_rate: 采样率
+            waveform_params: 波形参数 {
+                'type': 'LFM'/'NLFM'/'PhaseCode',
+                'bandwidth': 带宽(Hz),
+                'duration': 持续时间(s)
+            }
+        """
+        t = np.linspace(0, waveform_params['duration'], int(sample_rate * waveform_params['duration']))
+        
+        if waveform_params['type'] == 'LFM':
+            waveform = signal.chirp(t, f0=0, f1=waveform_params['bandwidth'], t1=waveform_params['duration'])
+        elif waveform_params['type'] == 'NLFM':
+            # 非线性调频波形
+            beta = 2.5
+            waveform = np.cos(2*np.pi* waveform_params['bandwidth'] * (t + beta*t**3))
+        elif waveform_params['type'] == 'PhaseCode':
+            # 相位编码波形
+            code_length = 13
+            phase_code = np.exp(1j*np.pi*np.random.randint(0,2,code_length))
+            waveform = np.kron(phase_code, np.ones(int(sample_rate*waveform_params['duration']/code_length)))
+        else:
+            raise ValueError("不支持的波形类型")
+        
+        return tx_signal * waveform[:len(tx_signal)]
+
+
+        

algo/jamming_signal_algo.py

@@ -5,36 +5,36 @@ class JammingSignalAlgo:
     @staticmethod
     def frequency_modulation(bandwidth, duration, sample_rate, modulation_index=0.5):
         """噪声调频干扰算法"""
-        t = np.arange(0, duration, 1/sample_rate)
-        baseband = np.random.normal(0, 1, len(t))
+        t = np.linspace(0, duration, 100)
+        baseband = np.random.normal(0, 1, 100)
         modulated = np.sin(2 * np.pi * bandwidth * t + modulation_index * np.cumsum(baseband))
-        return modulated
+        return modulated, -1.0, 1.0
 
     @staticmethod
     def repeater_jamming(bandwidth, duration, sample_rate, delay=0.1, attenuation=0.5):
         """转发式干扰算法"""
-        t = np.arange(0, duration, 1/sample_rate)
+        t = np.linspace(0, duration, 100)
         original_signal = np.sin(2 * np.pi * bandwidth * t)
-        return attenuation * np.roll(original_signal, int(delay * sample_rate))
+        return attenuation * np.roll(original_signal, int(delay * sample_rate)), -attenuation, attenuation
 
     @staticmethod
     def velocity_deception(bandwidth, duration, sample_rate, doppler_shift=1000):
         """速度欺骗干扰算法"""
-        t = np.arange(0, duration, 1/sample_rate)
-        return np.sin(2 * np.pi * (bandwidth + doppler_shift) * t)
+        t = np.linspace(0, duration, 100)
+        return np.sin(2 * np.pi * (bandwidth + doppler_shift) * t), -1.0, 1.0
 
     @staticmethod
     def range_deception(bandwidth, duration, sample_rate, fake_range=500):
         """距离欺骗干扰算法"""
-        t = np.arange(0, duration, 1/sample_rate)
+        t = np.linspace(0, duration, 100)
         fake_delay = 2 * fake_range / 3e8  # 计算双程时延
-        return np.sin(2 * np.pi * bandwidth * (t - fake_delay))
+        return np.sin(2 * np.pi * bandwidth * (t - fake_delay)), -1.0, 1.0
 
     @staticmethod
     def amplitude_modulation(bandwidth, duration, sample_rate, modulation_depth=0.8):
         """噪声调幅干扰算法"""
-        t = np.arange(0, duration, 1/sample_rate)
+        t = np.linspace(0, duration, 100)
         carrier = np.cos(2 * np.pi * bandwidth * t)
-        noise = np.random.normal(0, 1, len(t))
+        noise = np.random.uniform(-1, 1, 100)
         modulated = (1 + modulation_depth * noise) * carrier
-        return modulated
+        return modulated, -(1.0 + modulation_depth), 1.0 + modulation_depth

config.ini

@@ -1,3 +1,3 @@
 [NACOS]
-NACOS_SERVER_ADDR = localhost:8848
+NACOS_SERVER_ADDR = 47.92.123.216:8848
 SERVICE_IP = 192.168.5.5

model/jammer_radar.py

@@ -1,34 +1,23 @@
+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
 
-    def generate_jamming_signal(self, algorithm: callable, bandwidth: float, sample_rate: float, duration: float, gain: float) -> np.ndarray:
-        """
-        自主生成干扰信号
-        :param algorithm: 干扰算法
-        :param bandwidth: 信号带宽 (Hz)
-        :param sample_rate: 采样率 (Hz)
-        :param duration: 信号持续时间 (秒)
-        :param gain: 增益 (dB)
-        """
-        # 生成基带噪声信号
-        num_samples = int(duration * sample_rate)
-        base_signal = np.random.randn(num_samples) * 0.1 * gain
-        
+    def generate_jamming_signal(self, algorithm: callable, bandwidth: float, sample_rate: float, duration: float, gain: float) -> Tuple[np.ndarray, float, float]:
         # 应用干扰算法
-        jamming_signal = algorithm(
-            base_signal=base_signal,
+        signal, lower_bound, upper_bound, *_ = algorithm(  # 使用*_接收多余返回值
             bandwidth=bandwidth,
+            duration=duration,
             sample_rate=sample_rate
         )
-        return jamming_signal.astype(np.complex64)
+        return signal, lower_bound, upper_bound
 
-    def execute_jamming(self, algorithm: callable, bandwidth: float, sample_rate: float, duration: float, gain: float) -> np.ndarray:
+    def execute_jamming(self, algorithm: callable, bandwidth: float, sample_rate: float, duration: float, gain: float) -> Tuple[np.ndarray, float, float]:
         """
         执行自主干扰生成
         :param algorithm: 干扰算法
@@ -57,6 +46,17 @@ class JammerRadar:
         print('干扰雷达已接收信号')
         return rx_signal
 
+    def send_jamming_signal(self, algorithm, bandwidth, sample_rate, duration, 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
+
 
 
 

model/surveillance_radar.py

@@ -1,5 +1,6 @@
 import numpy as np
 from scipy.signal import find_peaks
+from algo.anti_jamming_signal_algo import AntiJammingSignalAlgo
 # ==================== 侦查雷达类 ====================
 class SurveillanceRadar:
     #初始化函数，传入usrp，输入的通道，输出的通道
@@ -28,7 +29,21 @@ class SurveillanceRadar:
         :param algorithm: 发射端抗干扰算法
         :param sample_rate: 采样率
         """
-        return algorithm(tx_signal=tx_signal, sample_rate=sample_rate)
+        
+        #如果为频率捷变算法
+        if algorithm == AntiJammingSignalAlgo.frequency_agility:
+            hop_sequence = [(100e6, 0.001), (200e6, 0.001)]  # 跳频序列，频率点和驻留时间
+            processed_signal = algorithm(tx_signal=tx_signal, sample_rate=sample_rate, hop_sequence=hop_sequence)
+            
+        
+        if algorithm == AntiJammingSignalAlgo.waveform_agility:
+            waveform_params = {'type': 'LFM', 'bandwidth': 100e6, 'duration': 0.1}  # 波形参数
+            processed_signal = algorithm(tx_signal=tx_signal, sample_rate=sample_rate, waveform_params=waveform_params)
+            
+        print("processed_signal:", processed_signal)
+        return processed_signal.astype(np.complex64)
+        
+        
 
     def apply_anti_jamming_processing(self, rx_signal: np.ndarray, algorithm: callable, sample_rate: float, **kwargs) -> np.ndarray:
         """

service/service.py

@@ -72,150 +72,126 @@ class Service:
             if Service.status == 0:
                 return ResponseDTO.ERROR_MS_DATA('SDR设备异常',
                                                  {"status": Service.status, "Error": "SDR设备未初始化"}).to_json()
-            try:
-                # 使用已初始化的雷达实例
-                surveillance_radar = Service.surveillance_radar
-                jammer_radar = Service.jammer_radar
-
-                jamming_policy = payload.get('jamming_policy')
-                anti_jamming_policy = payload.get('anti_jamming_policy')
-                # 判断策略是否合法
-                if not JammingPolicy.is_valid(jamming_policy) or not AntiJammingPolicy.is_valid(anti_jamming_policy):
-                    return ResponseDTO.ERROR_MS_DATA('策略不合法', {"status": Service.status, "Error": "策略不合法"}).to_json()
-                # 打印对应策略
-                print(' jamming_policy:', jamming_policy)
-                print(' anti_jamming_policy:', anti_jamming_policy)
-                # 根据策略选择算法
-                jam_algorithm = Service._get_jamming_algorithm_method(jamming_policy)
-                anti_jam_algorithm = Service._get_anti_jamming_algorithm(anti_jamming_policy, 'rx')
-
-                # 1. 应用发射端抗干扰算法（波形/频率捷变）
-                tx_original = np.random.randn(100)
-                if anti_jamming_policy in {AntiJammingPolicy.WAVEFORM_AGILITY, AntiJammingPolicy.FREQUENCY_AGILITY}:
-                    tx_processed = surveillance_radar.process_transmit_signal(
-                        tx_signal=tx_original,
-                        algorithm=Service._get_tx_algorithm_method(anti_jamming_policy.value),
-                        sample_rate=1e6
-                    )
-                else:
-                    tx_processed = tx_original
-
-                # 2. 发送经过处理的发射信号
-                tx_signal = surveillance_radar.send_signal(
-                    tx_signal=tx_processed,
-                    duration=0.1,
-                    center_freq=100e6,
-                    sample_rate=1e6,
-                    gain=20
+            # try:
+            # 使用已初始化的雷达实例
+            surveillance_radar = Service.surveillance_radar
+            jammer_radar = Service.jammer_radar
+
+            jamming_policy = payload.get('jamming_policy')
+            anti_jamming_policy = payload.get('anti_jamming_policy')
+            # 判断策略是否合法
+            if not JammingPolicy.is_valid(jamming_policy) or not AntiJammingPolicy.is_valid(anti_jamming_policy):
+                return ResponseDTO.ERROR_MS_DATA('策略不合法', {"status": Service.status, "Error": "策略不合法"}).to_json()
+            # 打印对应策略
+            print(' jamming_policy:', jamming_policy)
+            print(' anti_jamming_policy:', anti_jamming_policy)
+            # 根据策略选择算法
+            jam_algorithm = Service._get_jamming_algorithm_method(jamming_policy)
+            anti_jam_algorithm = Service._get_anti_jamming_algorithm(anti_jamming_policy)
+            print(' jamming_algorithm:', jam_algorithm)
+            print(' anti_jamming_algorithm:', anti_jam_algorithm)
+
+            # 1. 应用发射端抗干扰算法（波形/频率捷变）
+            tx_original = np.random.randn(100)
+            if anti_jamming_policy in {AntiJammingPolicy.WAVEFORM_AGILITY.value, AntiJammingPolicy.FREQUENCY_AGILITY.value}:
+                tx_processed = surveillance_radar.process_transmit_signal(
+                    tx_signal=tx_original,
+                    algorithm= anti_jam_algorithm,
+                    sample_rate=1e6
                 )
-
-                # 3. 干扰雷达生成干扰
-                jammed_signal = jammer_radar.execute_jamming(
-                    algorithm=jam_algorithm,
-                    bandwidth=100e6,
-                    sample_rate=1e6,
-                    duration=0.1,
-                    gain=20
-                ) if jam_algorithm else tx_signal
-
-                # 4. 干扰雷达发送干扰信号
-                jammer_radar.send_signal(
-                    tx_signal=jammed_signal,
+            else:
+                tx_processed = tx_original
+
+            # 生成干扰信号
+            jammed_signal, lower_bound, upper_bound = jammer_radar.execute_jamming(
+                algorithm=jam_algorithm,
+                bandwidth=100e6,
+                sample_rate=1e6,
+                duration=0.1,
+                gain=20
+            ) 
+
+            # 2.发送侦察信号
+            import time
+            from threading import Event
+            signal_sent_event = Event()
+            
+            def send_signal_thread():
+                tx_processed_signal = surveillance_radar.send_signal(
+                    tx_signal=tx_processed,
                     duration=0.1,
                     center_freq=100e6,
                     sample_rate=1e6,
                     gain=20
                 )
-
-                # 5. 侦查雷达接收信号
-                rx_original = surveillance_radar.recv_signal(
-                    num_samples=100,
+                signal_sent_event.set()
+            
+            import threading
+            send_thread = threading.Thread(target=send_signal_thread)
+            send_thread.start()
+            
+            # 5. 侦查雷达接收信号
+            # 等待信号发送完成后再接收
+            signal_sent_event.wait()
+            rx_original = surveillance_radar.recv_signal(
+                num_samples=100,
+                sample_rate=1e6,
+                center_freq=100e6
+            )
+
+            # 6. 将接收的原始信号与干扰信号结合
+            combined_signal = rx_original + jammed_signal
+            # 7. 接收端抗干扰处理（仅限接收端策略）
+            processed_signal = combined_signal
+            if anti_jamming_policy in {AntiJammingPolicy.POLARIZATION_FILTER.value} :
+                processed_signal = surveillance_radar.apply_anti_jamming_processing(
+                    rx_signal=combined_signal,
+                    algorithm=anti_jam_algorithm,
                     sample_rate=1e6,
-                    center_freq=100e6
+                    polarization_params={'angle': 45, 'ellipticity': 0.5}
                 )
 
-                # 6.接收端抗干扰处理（仅限接收端策略）
-                processed_signal = rx_original
-                if anti_jam_algorithm:
-                    processed_signal = surveillance_radar.execute_anti_jamming(
-                        rx_signal=rx_original,
-                        algorithm=anti_jam_algorithm,
-                        sample_rate=1e6,
-                        polarization_params={'angle': 45, 'ellipticity': 0.5}
-                    )
-
-                # 类型检查并转换复数信号为实数
-                if np.iscomplexobj(processed_signal):
-                    processed_signal = np.real(processed_signal)
-                if np.iscomplexobj(jammed_signal):
-                    jammed_signal = np.real(jammed_signal)
-
-                # 返回结果
-                return ResponseDTO.SUCCESS({"jamming_signal": np.real(jammed_signal).tolist(),
-                                            "anti_jamming_signal": np.real(processed_signal).flatten().tolist()}).to_json()
-            except Exception as e:
-                print('数据处理异常', e)
-                return ResponseDTO.ERROR_MS_DATA('数据处理异常', {"status": Service.status, "Error": str(e)}).to_json()
+            # 类型检查并转换复数信号为实数
+            if np.iscomplexobj(processed_signal):
+                processed_signal = np.real(processed_signal)
+            if np.iscomplexobj(jammed_signal):
+                jammed_signal = np.real(jammed_signal)
+
+            # 返回结果
+            return ResponseDTO.SUCCESS({
+                    "jamming_signal": np.real(jammed_signal).tolist(),
+                    "anti_jamming_signal": np.real(processed_signal).flatten().tolist(),
+                    "jamming_range": {
+                        "upper": upper_bound,
+                        "lower": lower_bound
+                    }
+                }).to_json()
+            # except Exception as e:
+            #     print('数据处理异常', e)
+            #     return ResponseDTO.ERROR_MS_DATA('数据处理异常', {"status": Service.status, "Error": str(e)}).to_json()
 
     @staticmethod
-    def send():
-        try:
-            # 设置中心频率、采样率和增益
-            center_freq = 100e6  # 2.4 GHz
-            sample_rate = 1e6  # 1 MS/s
-            duration = 10  # 以秒为单位
-            gain = 20  # [dB] 建议一开始设置小一点，按照实际情况调整
-            # 生成发送信号
-            num_samples = 100
-            tx_signal = np.random.randn(num_samples) + 0.1j * np.random.randn(num_samples)  # 修复部分
-
-            # 发送信号
-            Service.usrp.send_waveform(tx_signal, duration, center_freq, sample_rate, [0], gain)
-
-            # 接收信号
-            rx_signal = Service.usrp.recv_num_samps(num_samples, center_freq, sample_rate)
-            print('信号已发送:')
-            print(rx_signal)
-            print('信号已接收')
-        except Exception as e:
-            print('发送或接收信号异常', e)
-            Service.status = 0  # 发送或接收信号失败，状态置为0
-            Service.usrp = None  # 重置USRP对象
-    @staticmethod
     def _get_jamming_algorithm_method(policy: str) -> callable:
         # 干扰算法映射
         jam_algorithm_map = {
-            JammingPolicy.FREQUENCY_MODULATION: JammingSignalAlgo.frequency_modulation,
-            JammingPolicy.AMPLITUDE_MODULATION: JammingSignalAlgo.amplitude_modulation,
-            JammingPolicy.REPEATER: JammingSignalAlgo.repeater_jamming,
-            JammingPolicy.VELOCITY_DECEPTION: JammingSignalAlgo.velocity_deception,
-            JammingPolicy.RANGE_DECEPTION: JammingSignalAlgo.range_deception
+            JammingPolicy.FREQUENCY_MODULATION.value: JammingSignalAlgo.frequency_modulation,
+            JammingPolicy.AMPLITUDE_MODULATION.value: JammingSignalAlgo.amplitude_modulation,
+            JammingPolicy.REPEATER.value: JammingSignalAlgo.repeater_jamming,
+            JammingPolicy.VELOCITY_DECEPTION.value: JammingSignalAlgo.velocity_deception,
+            JammingPolicy.RANGE_DECEPTION.value: JammingSignalAlgo.range_deception
         }
         return jam_algorithm_map.get(policy)
 
     @staticmethod
-    def _get_anti_jamming_algorithm(policy: AntiJammingPolicy, algorithm_type: str) -> callable:
+    def _get_anti_jamming_algorithm(policy: str) -> callable:
         """
         统一抗干扰策略路由方法
         :param policy: 策略名称
-        :param algorithm_type: 算法类型['tx'发射端/'rx'接收端]
         """
-        tx_algorithms = {
-            AntiJammingPolicy.WAVEFORM_AGILITY: AntiJammingSignalAlgo.adaptive_filter,
-            AntiJammingPolicy.FREQUENCY_AGILITY: AntiJammingSignalAlgo.time_frequency_filter
-        }
-        rx_algorithms = {
-            AntiJammingPolicy.POLARIZATION_FILTER: AntiJammingSignalAlgo.polarization_filter
+        anti_jam_algorithm_map = {
+            AntiJammingPolicy.WAVEFORM_AGILITY.value: AntiJammingSignalAlgo.waveform_agility,
+            AntiJammingPolicy.FREQUENCY_AGILITY.value: AntiJammingSignalAlgo.frequency_agility,
+            AntiJammingPolicy.POLARIZATION_FILTER.value: AntiJammingSignalAlgo.polarization_filter
         }
         
-        if algorithm_type == 'tx':
-            return tx_algorithms.get(policy)
-        elif algorithm_type == 'rx':
-            return None if policy in {AntiJammingPolicy.WAVEFORM_AGILITY, AntiJammingPolicy.FREQUENCY_AGILITY} else rx_algorithms.get(policy)
-
-
-# main方法
-if __name__ == '__main__':
-    Service.initialize_usrp()
-
-
+        return anti_jam_algorithm_map.get(policy)

test.py

@@ -0,0 +1,35 @@
+import requests
+from common.strategy_enum import JammingPolicy, AntiJammingPolicy
+
+server_url = "http://localhost:5000"  # 替换为你的服务器地址和端口号
+
+# 初始化设备
+initialize_url = f"{server_url}/api/sdk/initialize"
+response = requests.get(initialize_url,timeout=60)
+if response.status_code == 200 and response.json()["code"] == 200:
+    print("设备初始化成功")
+else:
+    print("设备初始化失败"+str(response.json()))
+    exit()
+
+# 笛卡尔积测试数据获取接口
+data_url = f"{server_url}/api/sdk/data"
+jamming_policies = [policy.value for policy in JammingPolicy]  # 获取所有干扰策略
+anti_jamming_policies = [policy.value for policy in AntiJammingPolicy]  # 获取所有抗干扰策略
+
+# 生成笛卡尔积测试数据
+test_data = []
+for jamming_policy in jamming_policies:
+    for anti_jamming_policy in anti_jamming_policies:
+        test_data.append({"jamming_policy": jamming_policy, "anti_jamming_policy": anti_jamming_policy})
+print("笛卡尔积测试数据：", test_data)
+print("------------------------")
+# 发送测试数据并获取结果
+for data in test_data:
+    response = requests.post(data_url, json=data)
+    if response.status_code == 200 and response.json()["code"] == 200:
+        print(f"测试数据: {data}")
+        print(f"结果: {response.json()['data']}")
+        print("------------------------")
+    else:
+        print("失败：", response.json())