import json from math import e from re import S from matplotlib.colors import CenteredNorm import uhd import numpy as np import threading from dto.response_dto import ResponseDTO from model.surveillance_radar import SurveillanceRadar from model.jammer_radar import JammerRadar from algo.jamming_signal_algo import JammingSignalAlgo from algo.anti_jamming_signal_algo import AntiJammingSignalAlgo # 定义一组通道常量 CHANNEL_1 = 0 CHANNEL_2 = 1 CHANNEL_3 = 2 CHANNEL_4 = 3 SAMPLING_RATE = 20e6 CENTER_FREQ = 6e9 GAIN = 50 # 定义干扰、抗干扰策略集合，确保集合中的元素为合法的字符串常量 from common.strategy_enum import JammingPolicy, AntiJammingPolicy class Service: usrp = None # 静态类变量 status = 0 # 静态类变量 surveillance_radar = None # 侦查雷达实例 jammer_radar = None # 干扰雷达实例 _rlock = threading.RLock() # 可重入锁 @staticmethod def initialize_usrp(): with Service._rlock: Service.get_sdr_status() if Service.status == 1: print('USRP设备已经初始化') return ResponseDTO.SUCCESS({"status": Service.status}).to_json() try: Service.usrp = uhd.usrp.MultiUSRP() print('------SDR Devices initialize success！------') #侦查雷达初始化 Service.surveillance_radar = SurveillanceRadar(Service.usrp, rx_channel=CHANNEL_1, tx_channel=CHANNEL_1) #侦察类雷达启动 Service.surveillance_radar.run() # 干扰雷达初始化 Service.jammer_radar = JammerRadar(Service.usrp, rx=CHANNEL_2, tx=CHANNEL_2) Service.status = 1 return ResponseDTO.SUCCESS({"status": Service.status}).to_json() except Exception as e: print('SDR设备异常', e) Service.usrp = uhd.usrp.MultiUSRP("type=x4xx,reset") Service.status = 0 # 初始化失败，状态置为0 return ResponseDTO.ERROR_MS_DATA('SDR设备异常', {"status": Service.status, "Error": str(e)}).to_json() # @staticmethod # def get_sdr_status(): # with Service._rlock: # if Service.status == 0: # return ResponseDTO.ERROR_MS_DATA('SDR设备异常',{"status": Service.status, "Error": "SDR设备未初始化"}).to_json() # try: # rx_buffer = np.zeros((500000,), dtype=np.complex64) # Service.surveillance_radar.rx_stream.recv(rx_buffer, uhd.types.RXMetadata()) # Service.status = 1 # return ResponseDTO.SUCCESS({"status": Service.status}).to_json() # except Exception as e: # print('SDR设备异常', e) # Service.status = 0 # 获取状态失败，状态置为0 # Service.usrp = None # 重置USRP对象 # return ResponseDTO.ERROR_MS_DATA('SDR设备异常', {"status": Service.status, "Error": str(e)}).to_json() @staticmethod def get_sdr_status(): with Service._rlock: return ResponseDTO.SUCCESS({"status": Service.status}).to_json() @staticmethod def data(payload): with Service._rlock: 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) print(' jamming_algorithm:', jam_algorithm) print(' anti_jamming_algorithm:', anti_jam_algorithm) # 生成干扰信号 jammed_signal, lower_bound, upper_bound = jammer_radar.execute_jamming( algorithm=jam_algorithm, bandwidth=CENTER_FREQ, sample_rate=SAMPLING_RATE, duration=0.1, gain=GAIN ) # 抗干扰算法 、 干扰信号注入到侦察雷达 surveillance_radar.inject(anti_jam_algorithm,jammed_signal) # 返回结果 return ResponseDTO.SUCCESS({ "jamming_signal": np.real(jammed_signal).tolist(), # "anti_jamming_signal": np.real(surveillance_radar.processed_signal).flatten().tolist(), "anti_jamming_signal": surveillance_radar.processed_signal.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 _get_jamming_algorithm_method(policy: str) -> callable: # 干扰算法映射 jam_algorithm_map = { 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: str) -> callable: """ 统一抗干扰策略路由方法 :param policy: 策略名称 """ 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 } return anti_jam_algorithm_map.get(policy)