SDR/demo.py

import uhd
import numpy as np
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
from queue import Queue, Full
from threading import Thread, Event
import argparse
import time

# 确保中文显示正常


class USRPTransceiver:
    def __init__(self, args):
        self.args = args
        self.usrp = None
        self.tx_streamer = None
        self.rx_streamer = None
        self.stop_event = Event()

        # 固定长度缓冲队列 (300个信号)
        self.rx_signal_queue = Queue(maxsize=300)
        self.tx_signal_queue = Queue(maxsize=300)

        self.total_tx_signals = 0  # 总发送信号数
        self.total_rx_signals = 0  # 总接收信号数

        # 可视化相关
        self.fig, (self.ax_tx, self.ax_rx) = plt.subplots(2, 1, figsize=(10, 8))
        self.tx_line, = self.ax_tx.plot([], [], 'bo-', lw=1, label='发送信号强度')
        self.rx_line, = self.ax_rx.plot([], [], 'ro-', lw=1, label='接收信号强度')

        # 设置图表
        self.ax_tx.set_title('发送信号强度 (中间100个)')
        self.ax_rx.set_title('接收信号强度 (中间100个)')
        self.ax_tx.set_xlabel('信号序号')
        self.ax_rx.set_xlabel('信号序号')
        self.ax_tx.set_ylabel('信号强度')
        self.ax_rx.set_ylabel('信号强度')
        self.ax_tx.grid(True)
        self.ax_rx.grid(True)
        self.ax_tx.legend()
        self.ax_rx.legend()
        self.fig.tight_layout()

    def connect(self):
        """连接并配置USRP设备"""
        try:
            device_args = f"addr={self.args.ip},type=x4xx"
            self.usrp = uhd.usrp.MultiUSRP(device_args)
            print(f"成功连接到设备: {self.usrp.get_pp_string()}")

            # 配置发送端
            self.usrp.set_tx_rate(self.args.rate, self.args.channel)
            self.usrp.set_tx_freq(uhd.types.TuneRequest(self.args.freq), self.args.channel)
            self.usrp.set_tx_gain(self.args.gain, self.args.channel)
            self.usrp.set_tx_antenna("TX/RX", self.args.channel)

            # 配置接收端
            self.usrp.set_rx_rate(self.args.rate, self.args.channel)
            self.usrp.set_rx_freq(uhd.types.TuneRequest(self.args.freq), self.args.channel)
            self.usrp.set_rx_gain(self.args.gain, self.args.channel)
            self.usrp.set_rx_antenna("RX2", self.args.channel)

            # 打印实际配置
            print(f"实际采样率: TX = {self.usrp.get_tx_rate(self.args.channel) / 1e6} MHz, "
                  f"RX = {self.usrp.get_rx_rate(self.args.channel) / 1e6} MHz")
            print(f"实际中心频率: TX = {self.usrp.get_tx_freq(self.args.channel) / 1e9} GHz, "
                  f"RX = {self.usrp.get_rx_freq(self.args.channel) / 1e9} GHz")
            print(f"实际增益: TX = {self.usrp.get_tx_gain(self.args.channel)} dB, "
                  f"RX = {self.usrp.get_rx_gain(self.args.channel)} dB")

        except Exception as e:
            print(f"设备连接错误: {e}")
            raise

    def setup_streams(self):
        """设置发送和接收流"""
        # 设置发送流
        tx_stream_args = uhd.usrp.StreamArgs("fc32", "sc16")
        tx_stream_args.channels = [self.args.channel]
        self.tx_streamer = self.usrp.get_tx_stream(tx_stream_args)

        # 设置接收流
        rx_stream_args = uhd.usrp.StreamArgs("fc32", "sc16")
        rx_stream_args.channels = [self.args.channel]
        self.rx_streamer = self.usrp.get_rx_stream(rx_stream_args)

    def generate_signal(self, signal_id):
        """生成具有随机强度的测试信号"""
        t = np.arange(int(self.args.rate * 0.1)) / self.args.rate  # 0.1秒的数据

        # 基础幅度为0.5，添加随机波动 (±0.3)
        amplitude = 0.5 + 0.3 * np.random.randn()
        amplitude = max(0.1, min(0.9, amplitude))  # 限制在合理范围内

        # 添加随机相位偏移
        phase = np.random.uniform(0, 2 * np.pi)

        # 添加随机频率偏移 (±50kHz)
        freq_offset = np.random.uniform(-50e3, 50e3)

        signal = amplitude * np.exp(2j * np.pi * (1e6 + freq_offset) * t + 1j * phase)
        return signal.astype(np.complex64), amplitude

    def tx_worker(self):
        """发送线程函数"""
        try:
            tx_metadata = uhd.types.TXMetadata()
            tx_metadata.start_of_burst = True
            tx_metadata.end_of_burst = False
            tx_metadata.has_time_spec = False

            print("发送线程已启动")

            while not self.stop_event.is_set():
                # 生成具有随机强度的信号
                signal_id = self.total_tx_signals + 1
                signal, amplitude = self.generate_signal(signal_id)

                # 发送信号
                self.tx_streamer.send(signal, tx_metadata)

                # 将信号强度加入队列 (ID, 强度)
                try:
                    self.tx_signal_queue.put((signal_id, amplitude), block=False)
                except Full:
                    # 队列已满，移除最早的元素
                    self.tx_signal_queue.get()
                    self.tx_signal_queue.put((signal_id, amplitude))

                self.total_tx_signals += 1
                print(f"\r已发送信号: {self.total_tx_signals}", end="")

                # 控制发送频率
                time.sleep(0.05)

            # 发送结束标记
            tx_metadata.end_of_burst = True
            self.tx_streamer.send(np.zeros((1,), dtype=np.complex64), tx_metadata)
            print("\n发送线程已停止")

        except Exception as e:
            print(f"发送线程错误: {e}")

    def rx_worker(self):
        """接收线程函数"""
        try:
            # 启动接收流
            stream_cmd = uhd.types.StreamCMD(uhd.types.StreamMode.start_cont)
            stream_cmd.stream_now = True
            self.rx_streamer.issue_stream_cmd(stream_cmd)

            buffer_size = 500000
            rx_buffer = np.zeros((buffer_size,), dtype=np.complex64)
            rx_metadata = uhd.types.RXMetadata()

            print("接收线程已启动")
            overflow_count = 0

            while not self.stop_event.is_set():
                # 接收数据
                num_recv = self.rx_streamer.recv(rx_buffer, rx_metadata)

                if rx_metadata.error_code != uhd.types.RXMetadataErrorCode.none:
                    if rx_metadata.error_code == uhd.types.RXMetadataErrorCode.overflow:
                        overflow_count += 1
                        print(f"\r溢出错误计数: {overflow_count}", end="")
                    else:
                        print(f"接收错误: {rx_metadata.strerror()}")
                    print(f"接收错误: {rx_metadata.error_code}")
                    continue

                if num_recv > 0:
                    print(f"接收到的信号: 形状={rx_buffer.shape}, 类型={rx_buffer.dtype}")
                    # 计算接收信号的强度 (RMS)
                    signal_rms = np.sqrt(np.mean(np.abs(rx_buffer[:num_recv]) ** 2))
                    print(f"\r接收信号强度: {signal_rms:.3f}")
                    signal_id = self.total_rx_signals + 1

                    # 将信号强度加入队列 (ID, 强度)
                    try:
                        self.rx_signal_queue.put((signal_id, signal_rms), block=False)
                    except Full:
                        # 队列已满，移除最早的元素
                        self.rx_signal_queue.get()
                        self.rx_signal_queue.put((signal_id, signal_rms))

                    self.total_rx_signals += 1

            # 停止接收流
            stream_cmd = uhd.types.StreamCMD(uhd.types.StreamMode.stop_cont)
            self.rx_streamer.issue_stream_cmd(stream_cmd)
            print(f"\n接收线程已停止 - 溢出错误总数: {overflow_count}")

        except Exception as e:
            print(f"接收线程错误: {e}")

    def init_animation(self):
        """初始化动画"""
        self.ax_tx.set_xlim(0, 100)
        self.ax_rx.set_xlim(0, 100)
        self.ax_tx.set_ylim(0, 1.0)  # 强度范围
        self.ax_rx.set_ylim(0, 1.0)  # 强度范围
        return self.tx_line, self.rx_line

    def get_middle_signals(self, queue):
        """从队列中提取中间100个信号"""
        if queue.qsize() < 100:
            return []

        # 转换为列表
        items = list(queue.queue)

        # 计算中间100个元素的索引范围
        start_idx = max(0, len(items) // 2 - 50)
        end_idx = min(len(items), start_idx + 100)

        return items[start_idx:end_idx]

    def update_plot(self, frame):
        """更新绘图"""
        try:
            # 获取中间100个发送信号
            tx_signals = self.get_middle_signals(self.tx_signal_queue)
            if tx_signals:
                # 调整X轴为相对序号 (1-100)
                x = list(range(1, len(tx_signals) + 1))
                y = [item[1] for item in tx_signals]

                self.tx_line.set_data(x, y)
                self.ax_tx.set_xlim(1, len(tx_signals))
                self.ax_tx.set_ylim(max(0, min(y) - 0.05), max(y) + 0.05)

            # 获取中间100个接收信号
            rx_signals = self.get_middle_signals(self.rx_signal_queue)
            if rx_signals:
                # 调整X轴为相对序号 (1-100)
                x = list(range(1, len(rx_signals) + 1))
                y = [item[1] for item in rx_signals]

                self.rx_line.set_data(x, y)
                self.ax_rx.set_xlim(1, len(rx_signals))
                self.ax_rx.set_ylim(max(0, min(y) - 0.05), max(y) + 0.05)

            # 更新标题
            self.fig.suptitle(f"USRP信号强度监控 - 总发送: {self.total_tx_signals}, 总接收: {self.total_rx_signals}")

        except Exception as e:
            print(f"绘图更新错误: {e}")

        return self.tx_line, self.rx_line

    def run(self):
        """运行收发器"""
        try:

            self.connect()
            self.setup_streams()

            # 创建并启动线程
            tx_thread = Thread(target=self.tx_worker)
            rx_thread = Thread(target=self.rx_worker)

            tx_thread.start()
            rx_thread.start()

            # 启动实时绘图
            ani = FuncAnimation(self.fig, self.update_plot, init_func=self.init_animation,
                                interval=200, blit=True)

            plt.show()

            # 等待线程结束
            self.stop_event.set()
            tx_thread.join()
            rx_thread.join()

            print(f"完成收发 - 发送: {self.total_tx_signals}, 接收: {self.total_rx_signals}")

        except KeyboardInterrupt:
            print("用户中断，正在停止...")
            self.stop_event.set()
        finally:
            # 清理资源
            if self.usrp:
                self.usrp = None
            print("资源已清理")


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=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='通道号')

    args = parser.parse_args()

    transceiver = USRPTransceiver(args)
    transceiver.run()


if __name__ == "__main__":
    main()