commit

demo.py

@@ -291,7 +291,7 @@ class USRPTransceiver:
 def main():
     parser = argparse.ArgumentParser(description='USRP多线程信号收发器')
     parser.add_argument('--ip', default='192.168.10.2', help='USRP IP地址')
-    parser.add_argument('--rate', type=float, default=10e6, help='采样率 (Hz)')
+    parser.add_argument('--rate', type=float, default=5e6, help='采样率 (Hz)')
     parser.add_argument('--freq', type=float, default=6e9, help='中心频率 (Hz)')
     parser.add_argument('--gain', type=float, default=50, help='增益 (dB)')
     parser.add_argument('--channel', type=int, default=0, help='通道号')

model/surveillance_radar.py

@@ -13,7 +13,7 @@ import time
 import uhd
 # ==================== 侦查雷达类 ====================
 class SurveillanceRadar:
-    SAMPLING_RATE = 10e6
+    SAMPLING_RATE = 5e6
     CENTER_FREQ = 6e9
     GAIN = 50
 
@@ -117,7 +117,8 @@ class SurveillanceRadar:
         buff = np.zeros((self.samps_per_packet,), dtype=np.complex64)
         while not self.stop_event.is_set():
             try:
-                num_rx = self.rx_stream.recv(buff, metadata)
+                num_rx = self.rx_stream.recv(buff, metadata,timeout=10)
+                # time.sleep(0.5)
                 if metadata.error_code != uhd.types.RXMetadataErrorCode.none:
                     print(f"接收错误: {metadata.error_code}")
                     continue

test.py

@@ -12,24 +12,24 @@ 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]  # 获取所有抗干扰策略
+# # 笛卡尔积测试数据获取接口
+# 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())
+# # 生成笛卡尔积测试数据
+# 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())