first commit

algo/anti_jamming_singal_algo.py

@@ -0,0 +1,200 @@
+import numpy as np
+from scipy import signal, fft, linalg
+from typing import Tuple, List
+
+
+class AntiJammingSignalAlgo:
+    """
+    雷达抗干扰算法库
+    支持对抗噪声干扰、欺骗干扰、DRFM干扰等
+    """
+
+    @staticmethod
+    def adaptive_filter(
+            rx_signal: np.ndarray,
+            reference_signal: np.ndarray,
+            filter_length: int = 32,
+            mu: float = 0.01
+    ) -> np.ndarray:
+        """
+        LMS自适应滤波（对抗噪声干扰）
+        :param rx_signal: 含干扰的接收信号
+        :param reference_signal: 参考信号（干扰样本）
+        :param filter_length: 滤波器阶数
+        :param mu: 收敛因子 (0 < mu < 1)
+        :return: 滤波后信号
+        """
+        n = len(rx_signal)
+        w = np.zeros(filter_length, dtype=np.complex64)
+        output = np.zeros(n, dtype=np.complex64)
+
+        for k in range(filter_length, n):
+            x = reference_signal[k - filter_length:k][::-1]
+            output[k] = rx_signal[k] - np.dot(w, x)
+            w += mu * output[k] * np.conj(x)
+
+        return output
+
+    @staticmethod
+    def pca_denoise(
+            signal_matrix: np.ndarray,
+            n_components: int = 3
+    ) -> np.ndarray:
+        """
+        基于PCA的信号增强（对抗宽带噪声）
+        :param signal_matrix: 多通道信号矩阵 [n_channels, n_samples]
+        :param n_components: 保留的主成分数量
+        :return: 降噪后信号
+        """
+        cov_matrix = np.cov(signal_matrix)
+        eig_vals, eig_vecs = linalg.eigh(cov_matrix)
+        idx = eig_vals.argsort()[::-1]
+        principal_components = eig_vecs[:, idx[:n_components]]
+        return np.dot(principal_components.T, signal_matrix)
+
+    @staticmethod
+    def cfar_detection(
+            spectrum: np.ndarray,
+            guard_cells: int = 2,
+            training_cells: int = 10,
+            threshold_factor: float = 3.0
+    ) -> np.ndarray:
+        """
+        CFAR目标检测（对抗欺骗干扰）
+        :param spectrum: 距离/多普勒谱
+        :param guard_cells: 保护单元数
+        :param training_cells: 训练单元数
+        :param threshold_factor: 阈值系数
+        :return: 目标掩码（True表示目标）
+        """
+        mask = np.zeros_like(spectrum, dtype=bool)
+        n = len(spectrum)
+
+        for i in range(n):
+            left = max(0, i - guard_cells - training_cells)
+            right = min(n, i + guard_cells + training_cells)
+            noise_est = np.concatenate([
+                spectrum[left:i - guard_cells],
+                spectrum[i + guard_cells:right]
+            ])
+            threshold = threshold_factor * np.mean(noise_est)
+            mask[i] = spectrum[i] > threshold
+
+        return mask
+
+    @staticmethod
+    def time_frequency_filter(
+            rx_signal: np.ndarray,
+            sample_rate: float,
+            jamming_freq: float,
+            bandwidth: float = 1e6
+    ) -> np.ndarray:
+        """
+        时频联合滤波（对抗点频干扰）
+        :param rx_signal: 接收信号
+        :param sample_rate: 采样率
+        :param jamming_freq: 干扰中心频率
+        :param bandwidth: 抑制带宽
+        :return: 滤波后信号
+        """
+        # STFT参数
+        nperseg = 256
+        noverlap = nperseg // 2
+
+        # 计算STFT
+        f, t, Zxx = signal.stft(rx_signal, fs=sample_rate,
+                                nperseg=nperseg, noverlap=noverlap)
+
+        # 构建干扰掩码
+        jamming_mask = (np.abs(f - jamming_freq) < bandwidth / 2)
+
+        # 干扰置零
+        Zxx_clean = Zxx.copy()
+        Zxx_clean[jamming_mask, :] = 0
+
+        # 逆STFT
+        _, clean_signal = signal.istft(Zxx_clean, fs=sample_rate,
+                                       nperseg=nperseg, noverlap=noverlap)
+        return clean_signal[:len(rx_signal)]
+
+    @staticmethod
+    def mti_filter(
+            pulse_matrix: np.ndarray,
+            clutter_rejection: float = 30.0
+    ) -> np.ndarray:
+        """
+        动目标显示(MTI)滤波（对抗地物杂波）
+        :param pulse_matrix: 多脉冲信号矩阵 [n_pulses, n_samples]
+        :param clutter_rejection: 杂波抑制深度(dB)
+        :return: 滤波后信号矩阵
+        """
+        # 构造对消器 (3脉冲对消器)
+        canceler = np.array([1, -2, 1])
+        output = np.apply_along_axis(
+            lambda x: np.convolve(x, canceler, mode='valid'),
+            0, pulse_matrix
+        )
+        return output * (10 ** (clutter_rejection / 20))
+
+    @staticmethod
+    def spoofing_detection(
+            pulse_train: List[np.ndarray],
+            prf: float,
+            max_velocity: float = 300.0
+    ) -> np.ndarray:
+        """
+        欺骗干扰检测（基于多普勒连续性）
+        :param pulse_train: 脉冲序列
+        :param prf: 脉冲重复频率
+        :param max_velocity: 最大合理速度(m/s)
+        :return: 干扰标记数组
+        """
+        n_pulses = len(pulse_train)
+        lambd = 3e8 / 10e9  # 假设雷达波长(10GHz)
+        max_doppler = 2 * max_velocity / lambd
+
+        # 计算多普勒谱
+        doppler_spectra = []
+        for pulse in pulse_train:
+            spec = np.abs(fft.fftshift(fft.fft(pulse)))
+            doppler_spectra.append(spec)
+
+        # 检测异常多普勒跳变
+        is_spoofed = np.zeros(n_pulses, dtype=bool)
+        for i in range(1, n_pulses):
+            corr = np.correlate(doppler_spectra[i - 1], doppler_spectra[i], 'same')
+            peak_shift = np.argmax(corr) - len(corr) // 2
+            if abs(peak_shift) > max_doppler / prf * len(corr):
+                is_spoofed[i] = True
+
+        return is_spoofed
+
+
+# ==================== 使用示例 ====================
+if __name__ == "__main__":
+    # 模拟含干扰信号
+    fs = 100e6
+    t = np.arange(0, 1e-3, 1 / fs)
+    true_signal = np.exp(1j * 2 * np.pi * 1e6 * t)
+    noise_jam = 0.5 * (np.random.normal(0, 1, len(t)) + 1j * np.random.normal(0, 1, len(t)))
+    spoof_jam = 0.3 * np.exp(1j * 2 * np.pi * 3e6 * t)
+    rx_signal = true_signal + noise_jam + spoof_jam
+
+    # 示例1：自适应滤波
+    filtered = AntiJammingSignalAlgo.adaptive_filter(
+        rx_signal,
+        reference_signal=noise_jam,
+        filter_length=16,
+        mu=0.05
+    )
+    print(f"自适应滤波后SNR提升: {10 * np.log10(np.var(filtered) / np.var(rx_signal - filtered)):.1f} dB")
+
+    # 示例2：CFAR检测
+    spectrum = np.abs(fft.fft(rx_signal))
+    targets = AntiJammingSignalAlgo.cfar_detection(spectrum)
+    print(f"CFAR检测到目标位置: {np.where(targets)[0]}")
+
+    # 示例3：欺骗干扰检测
+    pulse_train = [rx_signal[i * 100:(i + 1) * 100] for i in range(10)]
+    spoof_flags = AntiJammingSignalAlgo.spoofing_detection(pulse_train, prf=1e3)
+    print(f"受干扰脉冲索引: {np.where(spoof_flags)[0]}")

algo/jamming_singal_algo.py

@@ -0,0 +1,189 @@
+import numpy as np
+from scipy.signal import chirp
+from typing import Union, List
+class JammingSignalAlgo:
+    """
+    雷达干扰信号生成算法库
+    支持噪声干扰、欺骗干扰、智能干扰等多种模式
+    """
+
+    @staticmethod
+    def generate_noise_jam(
+            bandwidth: float,
+            duration: float,
+            sample_rate: float,
+            jamming_type: str = "gaussian"
+    ) -> np.ndarray:
+        """
+        生成噪声干扰信号
+        :param bandwidth: 干扰带宽(Hz)
+        :param duration: 干扰持续时间(s)
+        :param sample_rate: 采样率(Hz)
+        :param jamming_type: 噪声类型(gaussian/blanket/spot)
+        :return: 复数干扰信号
+        """
+        nsamples = int(duration * sample_rate)
+
+        if jamming_type == "gaussian":
+            # 高斯白噪声（全频段均匀干扰）
+            return np.random.normal(0, 1, nsamples) + 1j * np.random.normal(0, 1, nsamples)
+
+        elif jamming_type == "blanket":
+            # 毯式噪声（带限增强）
+            noise = np.random.normal(0, 1, nsamples)
+            fft_sig = np.fft.fft(noise)
+            freqs = np.fft.fftfreq(nsamples, 1 / sample_rate)
+            mask = np.abs(freqs) <= bandwidth / 2
+            return np.fft.ifft(fft_sig * mask).astype(np.complex64)
+
+        elif jamming_type == "spot":
+            # 点频噪声（针对特定频点）
+            t = np.arange(nsamples) / sample_rate
+            carrier = np.exp(1j * 2 * np.pi * bandwidth / 2 * t)
+            return (np.random.normal(0, 1, nsamples) * carrier).astype(np.complex64)
+
+        else:
+            raise ValueError(f"未知噪声类型: {jamming_type}")
+
+    @staticmethod
+    def generate_deceptive_jam(
+            radar_params: dict,
+            false_targets: List[dict],
+            sample_rate: float
+    ) -> np.ndarray:
+        """
+        生成欺骗式干扰信号（假目标）
+        :param radar_params: 雷达参数 {
+            "pulse_width": 脉宽(s),
+            "bandwidth": 带宽(Hz),
+            "prf": 脉冲重复频率(Hz)
+        }
+        :param false_targets: 假目标参数列表 [{
+            "delay": 时延(s),
+            "doppler": 多普勒频移(Hz),
+            "amplitude": 幅度系数(0-1)
+        }]
+        :param sample_rate: 采样率(Hz)
+        :return: 复数干扰信号
+        """
+        pw = radar_params["pulse_width"]
+        bw = radar_params["bandwidth"]
+        nsamples = int(pw * sample_rate)
+
+        # 生成基础雷达信号（LFM）
+        t = np.linspace(0, pw, nsamples)
+        chirp_signal = chirp(t, f0=0, t1=pw, f1=bw, phi=90, method='linear')
+        chirp_signal = chirp_signal.astype(np.complex64)
+
+        # 合成假目标
+        jam_signal = np.zeros_like(chirp_signal)
+        for target in false_targets:
+            delay_samples = int(target["delay"] * sample_rate)
+            doppler_phase = 2 * np.pi * target["doppler"] * t
+            shifted_sig = np.roll(chirp_signal, delay_samples) * \
+                          np.exp(1j * doppler_phase) * \
+                          target["amplitude"]
+            jam_signal += shifted_sig[:len(chirp_signal)]
+
+        return jam_signal
+
+    @staticmethod
+    def generate_drgm_jam(
+            radar_signal: np.ndarray,
+            memory_depth: int = 5,
+            modulation: str = "amplitude"
+    ) -> np.ndarray:
+        """
+        数字射频存储器(DRGM)干扰
+        :param radar_signal: 截获的雷达信号
+        :param memory_depth: 存储深度（复制次数）
+        :param modulation: 调制方式(amplitude/phase/frequency)
+        :return: 干扰信号
+        """
+        jam_signal = np.array([])
+
+        for i in range(memory_depth):
+            if modulation == "amplitude":
+                mod_signal = radar_signal * (0.8 + 0.2 * np.random.rand())
+            elif modulation == "phase":
+                mod_signal = radar_signal * np.exp(1j * np.pi / 4 * i)
+            elif modulation == "frequency":
+                mod_signal = radar_signal * np.exp(1j * 2 * np.pi * 0.01 * i * np.arange(len(radar_signal)))
+            else:
+                mod_signal = radar_signal
+
+            jam_signal = np.concatenate([jam_signal, mod_signal])
+
+        return jam_signal
+
+    @staticmethod
+    def generate_smart_jam(
+            radar_signal: np.ndarray,
+            sample_rate: float,
+            jamming_mode: str = "repeater"
+    ) -> np.ndarray:
+        """
+        智能干扰（认知电子战）
+        :param radar_signal: 截获的雷达信号
+        :param sample_rate: 采样率
+        :param jamming_mode: 干扰模式(repeater/velocity_deception/range_deception)
+        :return: 干扰信号
+        """
+        if jamming_mode == "repeater":
+            # 转发式干扰（延迟转发）
+            return np.roll(radar_signal, int(0.1 * sample_rate))
+
+        elif jamming_mode == "velocity_deception":
+            # 速度欺骗（多普勒调制）
+            t = np.arange(len(radar_signal)) / sample_rate
+            return radar_signal * np.exp(1j * 2 * np.pi * 1e4 * t)
+
+        elif jamming_mode == "range_deception":
+            # 距离欺骗（时域复制）
+            return np.concatenate([
+                radar_signal[:len(radar_signal) // 2],
+                radar_signal[:len(radar_signal) // 2]
+            ])
+
+        else:
+            raise ValueError(f"未知智能干扰模式: {jamming_mode}")
+
+
+# ==================== 使用示例 ====================
+if __name__ == "__main__":
+    # 示例1：生成噪声干扰
+    noise_jam = JammingSignalAlgo.generate_noise_jam(
+        bandwidth=10e6,
+        duration=0.1,
+        sample_rate=100e6,
+        jamming_type="blanket"
+    )
+    print(f"噪声干扰信号长度：{len(noise_jam)} samples")
+
+    # 示例2：生成欺骗干扰
+    false_targets = [
+        {"delay": 2e-6, "doppler": 1e4, "amplitude": 0.7},
+        {"delay": 5e-6, "doppler": -5e3, "amplitude": 0.3}
+    ]
+    deceptive_jam = JammingSignalAlgo.generate_deceptive_jam(
+        radar_params={"pulse_width": 50e-6, "bandwidth": 100e6, "prf": 1e3},
+        false_targets=false_targets,
+        sample_rate=200e6
+    )
+    print(f"欺骗干扰信号峰值幅度：{np.max(np.abs(deceptive_jam)):.2f}")
+
+    # 示例3：DRGM干扰
+    radar_signal = np.exp(1j * 2 * np.pi * 1e6 * np.arange(1000) / 100e6)
+    drgm_jam = JammingSignalAlgo.generate_drgm_jam(
+        radar_signal,
+        memory_depth=3,
+        modulation="phase"
+    )
+    print(f"DRGM干扰信号维度：{drgm_jam.shape}")
+# 算法	处理增益	时延	适用场景
+# 自适应滤波	15-25 dB	O(N)	对抗噪声干扰
+# PCA降噪	10-20 dB	O(n³)	多通道系统
+# CFAR检测	-	        O(N)	欺骗干扰抑制
+# 时频滤波	20-40 dB	O(NlogN)点频干扰
+# MTI滤波	30-50 dB	O(N)	动目标检测
+# 欺骗检测	-	        O(N²)	DRFM干扰

common/http_status.py

@@ -0,0 +1,27 @@
+# 生成一个http响应码以及message的枚举类
+
+from enum import Enum
+
+class HTTPStatus(Enum):
+    OK = (200, "OK")
+    CREATED = (201, "Created")
+    ACCEPTED = (202, "Accepted")
+    NO_CONTENT = (204, "No Content")
+    BAD_REQUEST = (400, "Bad Request")
+    UNAUTHORIZED = (401, "Unauthorized")
+    FORBIDDEN = (403, "Forbidden")
+    NOT_FOUND = (404, "Not Found")
+    METHOD_NOT_ALLOWED = (405, "Method Not Allowed")
+    INTERNAL_SERVER_ERROR = (500, "Internal Server Error")
+    NOT_IMPLEMENTED = (501, "Not Implemented")
+    BAD_GATEWAY = (502, "Bad Gateway")
+    SERVICE_UNAVAILABLE = (503, "Service Unavailable")
+
+    def __init__(self, code, message):
+        self.code = code
+        self.message = message
+
+    def __str__(self):
+        return f"{self.code} {self.message}"
+
+

config/nacos.py

@@ -9,9 +9,9 @@ import asyncio
 NACOS_NAMESPACE_ID = 'fire'
 NACOS_USERNAME = 'nacos'
 NACOS_PASSWORD = 'nacos'
-NACOS_SERVER_ADDR = '192.168.31.66:8848'
+NACOS_SERVER_ADDR = '10.203.149.104:8848'
 SERVICE_NAME = 'SDR'
-SERVICE_IP = '192.168.31.98'
+SERVICE_IP = '10.203.149.104'
 SERVICE_PORT = 5000
 
 class NacosConfig:
@@ -64,14 +64,14 @@ class NacosConfig:
 
     # 异步发送心跳
     def send_heartbeat_in_thread(self,client, service_name, ip, port):
-        # loop = asyncio.new_event_loop()
-        # asyncio.set_event_loop(loop)
-        # loop.run_until_complete(self.send_heartbeat(client, service_name, ip, port))
-        def run():
-            try:
-                asyncio.run(self.send_heartbeat(client, service_name, ip, port))
-            except KeyboardInterrupt:
-                print("Heartbeat task interrupted")
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        loop.run_until_complete(self.send_heartbeat(client, service_name, ip, port))
+        # def run():
+        #     try:
+        #         asyncio.run(self.send_heartbeat(client, service_name, ip, port))
+        #     except KeyboardInterrupt:
+        #         print("Heartbeat task interrupted")
 
     async def send_heartbeat(self,client, service_name, ip, port):
         while True:

config/swagger.py

@@ -6,8 +6,7 @@ from controller.controller import ns as controller_ns
 class SwaggerConfig:
     def __init__(self):
         self.api_blueprint = Blueprint("open_api", __name__, url_prefix="/api")
-        self.api = Api(self.api_blueprint, version="1.0",
-                  prefix="/v1", title="OpenApi", description="The Open Api Service")
+        self.api = Api(self.api_blueprint, version="1.0", title="OpenApi", description="The Open Api Service")
         self.mount_ns();
 
     # 挂载命名空间

controller/controller.py

@@ -1,15 +1,41 @@
 # controller/controller.py
-from flask_restx import Namespace, Resource, Api
+from flask_restx import Namespace, Resource, Api, fields
 from service import service
 # 创建命名空间
 ns = Namespace('sdk', description='SDK API')
 
 #控制器层
-@ns.route('/send')
-class handler1(Resource):
-    @ns.doc(description='返回一个简单的问候')
+
+@ns.route('/initialize')
+class handler(Resource):
+    @ns.doc(description='半实物初始化')
+    def get(self):
+        """半实物初始化"""
+        return service.initialize_usrp()
+
+@ns.route('/status')
+class handler(Resource):
+    @ns.doc(description='半实物状态获取')
     def get(self):
-        """返回一个简单的问候"""
-        return service.send()
+        """半实物状态获取"""
+        return service.getSDRStatus()
 
+@ns.route('/data')
+class handler(Resource):
+    @ns.doc(description='基于干扰策略、抗干扰策略获取数据')
+    @ns.expect(ns.model('/data-model', {
+    'jamming_policy': fields.String(required=True, description='干扰策略'),
+    'anti_jamming_policy': fields.String(required=True, description='抗干扰策略')
+    }))
+    def post(self):
+        """基于干扰策略、抗干扰策略获取数据"""
+        args = self.api.payload
+        return service.data(args)
+
+
+@ns.route('/demo')
+class handler(Resource):
+    @ns.doc(description='测试')
+    def get(self):
+        return service.demo()
 

dto/response_dto.py

@@ -0,0 +1,28 @@
+# 封装一个响应对象
+from common.http_status import HTTPStatus
+class ResponseDTO:
+    def __init__(self, code, message, data):
+        self.code = code
+        self.message = message
+        self.data = data
+    def to_json(self):
+        return {
+            "code": self.code,
+            "message": self.message,
+            "data": self.data
+        }
+    @staticmethod
+    def SUCCESS(data):
+        return ResponseDTO(HTTPStatus.OK.code, HTTPStatus.OK.message, data)
+
+    @staticmethod
+    def SUCCESS_MS_DATA(message, data):
+        return ResponseDTO(HTTPStatus.OK.code, message, data)
+
+    @staticmethod
+    def ERROR(data):
+        return ResponseDTO(HTTPStatus.INTERNAL_SERVER_ERROR.code, HTTPStatus.INTERNAL_SERVER_ERROR.message, data)
+
+    @staticmethod
+    def ERROR_MS_DATA(message, data):
+        return ResponseDTO(HTTPStatus.INTERNAL_SERVER_ERROR.code, message, data)

main.py

@@ -16,5 +16,4 @@ if __name__ == '__main__':
     # 注册蓝图
     app.register_blueprint(swagger_config.api_blueprint)
 
-
-    app.run(host=config.nacos.SERVICE_IP, port=config.nacos.SERVICE_PORT)
+    app.run(host='0.0.0.0', port=config.nacos.SERVICE_PORT)

model/jammer_radar.py

@@ -0,0 +1,29 @@
+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 send_signal(self, tx_signal, duration, center_freq, sample_rate, gain):
+        # 发送信号
+        self.usrp.send_waveform(tx_signal, duration, center_freq, sample_rate, self.tx, gain)
+        print('干扰雷达已发送信号')
+
+    # 封装一个接收信号的函数
+    def recv_signal(self, num_samples, sample_rate, center_freq):
+        rx_signal = self.usrp.recv_num_samps(num_samples, center_freq,sample_rate,self.rx);
+        print('侦查雷达已接收信号')
+        return rx_signal
+
+
+
+
+
+
+
+

model/surveillance_radar.py

@@ -0,0 +1,58 @@
+import numpy as np
+from scipy.signal import find_peaks
+# ==================== 侦查雷达类 ====================
+class SurveillanceRadar:
+    #初始化函数，传入usrp，输入的通道，输出的通道
+    def __init__(self, usrp, rx, tx):
+        self.usrp = usrp
+        self.rx = rx
+        self.tx = tx
+
+        # 封装一个发送信号的函数
+    def send_signal(self, tx_signal, duration, center_freq, sample_rate, gain):
+        # 发送信号
+        self.usrp.send_waveform(tx_signal, duration, center_freq, sample_rate, self.tx, gain)
+        print('侦查雷达已发送信号')
+
+    # 封装一个接收信号的函数
+    def recv_signal(self, num_samples, sample_rate, center_freq):
+        rx_signal = self.usrp.recv_num_samps(num_samples, center_freq, sample_rate, self.rx);
+        print('侦查雷达已接收信号')
+        return rx_signal
+    # 分析信号，获取目标距离
+    def analyze_signal(self, rx_signal: np.ndarray, sample_rate: float,
+                     cfar_threshold: float = 20.0) -> list:
+        """
+        分析接收信号并返回目标距离列表
+        :param rx_signal: 接收信号（复数形式）
+        :param sample_rate: 采样率（Hz）
+        :param cfar_threshold: CFAR检测阈值（dB）
+        :return: 目标距离列表（米）
+        """
+        # 1. 去斜处理（Dechirping）
+        mixed = rx_signal * np.conj(self.tx_signal[:len(rx_signal)])
+
+        # 2. 加窗处理（Hamming窗）
+        window = np.hamming(len(mixed))
+        windowed = mixed * window
+
+        # 3. 距离FFT
+        range_fft = np.fft.fft(windowed)
+        spectrum_db = 20 * np.log10(np.abs(range_fft) + 1e-10)  # 转换为dB
+
+        # 4. 计算距离轴
+        freq_bins = np.fft.fftfreq(len(spectrum_db), 1/sample_rate)
+        ranges = (self.c * self.T * freq_bins) / (2 * self.B)
+
+        # 5. CFAR目标检测（简化版）
+        noise_floor = np.median(spectrum_db)
+        peaks, _ = find_peaks(spectrum_db, height=noise_floor + cfar_threshold)
+
+        # 6. 提取目标距离（取前向部分）
+        valid_peaks = peaks[peaks <= len(ranges)//2]
+        print(f"检测到目标距离：{[abs(ranges[p]) for p in valid_peaks]} 米")
+        return [abs(ranges[p]) for p in valid_peaks]
+
+
+
+

service/service.py

@@ -1,36 +1,135 @@
+import json
+
 import uhd
 import numpy as np
 
-try:
-    # 创建USRP设备对象
-    usrp = uhd.usrp.MultiUSRP()
-    print('------SDR Devices initialize success！------')
-except RuntimeError as e:
-    print('SDR设备异常', e)
+from dto.response_dto import ResponseDTO
+from model.surveillance_radar import SurveillanceRadar
+from model.jammer_radar import JammerRadar
+# 初始化侦查雷达
+# surveillance_radar = SurveillanceRadar(usrp,rx=CHANNEL_1,tx=CHANNEL_1)
+# Jammer_radar = JammerRadar(usrp,rx=CHANNEL_2,tx=CHANNEL_2)
+# 定义一组通道常量
+CHANNEL_1 = 0
+CHANNEL_2 = 1
+CHANNEL_3 = 2
+CHANNEL_4 = 3
+
+# 定义干扰策略集合，确保集合中的元素为合法的字符串常量
+JAMMING_POLICY = {
+    "噪声调频", "噪声调幅", "噪声直放", "速度多假目标", "距离多假目标"
+}
+ANTI_JAMMING_POLICY = {
+    "频率捷变", "波形捷变", "自适应极化滤波"
+}
+
+# 定义一个USRP对象
+usrp = None
+
+# 初始化USRP设备对象
+def initialize_usrp():
+    # try:
+    #     usrp = uhd.usrp.MultiUSRP()
+    #     usrp.close()
+    #     print('------SDR Devices initialize success！------')
+    return ResponseDTO.SUCCESS({"status": 1}).to_json()
+    # except Exception as e:
+    #     print('SDR设备异常', e)
+    #     return ResponseDTO.ERROR_MS_DATA('SDR设备异常',{"status": 0,
+    #                              "Error": str(e)}).to_json()
+
+# 获取SDR状态
+def getSDRStatus():
+    # try:
+    #     samples = usrp.recv_num_samps(10000, 100e6, 1e6, [0], 50) # 单位: 需要接收的采样点总数（无单位）, Hz, Hz, channel IDs 的列表, dB
+    return ResponseDTO.SUCCESS({"status": 1}).to_json()
+    # except Exception as e:
+    #     print('SDR设备异常', e)
+    #     return ResponseDTO.ERROR_MS_DATA('SDR设备异常', {"status": 0,
+    #                              "Error": str(e)}).to_json()
+# 关闭USRP设备对象
+def close_usrp():
+    try:
+        usrp.close()
+        print('------SDR Devices close success！------')
+    except RuntimeError as e:
+        print('SDR设备异常', e)
+
+def data(payload):
+    jamming_policy = payload['jamming_policy']
+    anti_jamming_policy = payload['anti_jamming_policy']
+    # 判断策略是否合法
+    if jamming_policy not in JAMMING_POLICY or anti_jamming_policy not in ANTI_JAMMING_POLICY:
+        return ResponseDTO.ERROR_MS_DATA('策略不合法', {"status": 0,
+                                 "Error": "策略不合法"}).to_json()
+    #打印对应策略
+    print(' jamming_policy:', jamming_policy)
+    print(' anti_jamming_policy:', anti_jamming_policy)
+    #生成两个长度为一百的浮点数组，代表侦察信号和返回的信号，数组元素随机
+    surveillance_signal = np.random.randn(100);
+    surveillance_signal_back = np.random.randn(100);
+    #返回结果
+    return ResponseDTO.SUCCESS({"jamming_signal": surveillance_signal.tolist(),
+                                "anti_jamming_signal": surveillance_signal_back.tolist()}).to_json()
+
 
 def send():
+    # # 设置中心频率、采样率和增益
+    # center_freq = 100e6  # 2.4 GHz
+    # sample_rate = 1e6  # 1 MS/s
+    # duration = 10  # 以秒为单位
+    # gain = 20  # [dB] 建议一开始设置小一点，按照实际情况调整
+    # # 生成发送信号
+    # num_samples = 1
+    # tx_signal = np.random.randn(num_samples) + 0.1j * np.random.randn(num_samples)  # 修复部分
+    #
+    # # 发送信号
+    # usrp.send_waveform(tx_signal, duration, center_freq, sample_rate, [0], gain)
+    #
+    # # 接收信号
+    # rx_signal = usrp.recv_num_samps(num_samples, center_freq,sample_rate)
+    # print('信号已发送:')
+    # print(rx_signal)
+    # print('信号已接收')
+    # return {
+    #     "rx_signal": rx_signal.tolist()
+    # }
+    return {
+        "rx_signal": [1,2,3,4,5,6,7,8,9,10]
+    }
+
+def demo():
     # 设置中心频率、采样率和增益
     center_freq = 100e6  # 2.4 GHz
     sample_rate = 1e6  # 1 MS/s
     duration = 10  # 以秒为单位
     gain = 20  # [dB] 建议一开始设置小一点，按照实际情况调整
     # 生成发送信号
-    num_samples = 1000
+    num_samples = 100
     tx_signal = np.random.randn(num_samples) + 0.1j * np.random.randn(num_samples)  # 修复部分
 
     # 发送信号
     usrp.send_waveform(tx_signal, duration, center_freq, sample_rate, [0], gain)
 
     # 接收信号
-    rx_signal = usrp.recv_num_samps(num_samples, sample_rate,center_freq)
+    rx_signal = usrp.recv_num_samps(num_samples, center_freq,sample_rate)
     print('信号已发送:')
-    print(rx_signal)
+    # 将复数数组转换为字符串数组
+    rx_signal_str = [str(s) for s in rx_signal]
+    print(rx_signal_str)
     print('信号已接收')
+    return {
+        "rx_signal": rx_signal_str
+    }
+
+
+# 获取SDR状态,如果状态正常，则返回True，否则返回False
 
 
 #main方法
 if __name__ == '__main__':
-    send()
+    demo()
+
 
 
 

test.py

@@ -1,5 +0,0 @@
-
-import uhd
-usrp = uhd.usrp.MultiUSRP()
-samples = usrp.recv_num_samps(10000, 100e6, 1e6, [0], 50)
-print(samples[0:10])