目录

前言 

频谱仪简介

硬件接线图

实现原理

代码详解

 步骤1： 建立连接

步骤2： 发送SCPI指令

步骤3： 根据需求封装函数实现功能

项目源码

支持的设备信号

前言 

为什么要实现频谱仪的控制呢？起因是因为我司RF的同事们需要对通信模块进行GSM的带内杂散进行测试，由于该测试的频点涉及900多个，所以急需开发一款自动化测试工具。

该工具需同时实现对无线通讯仪和频谱仪的控制，这篇文章主要介绍一下频谱仪的控制！

频谱仪简介

做通信的应该对频谱仪都不陌生，不过我之前做的都是软件相关行业，所以我对频谱仪其实还是很陌生的；不过好在我的同时们都很热心，加上仪器的文档，一段时间下来，也算是熟悉了一些。

N9010B EXA 信号分析仪，分析的频域范围是10 Hz 至 44 GHz：

硬件接线图

实现原理

如何用python实现对频谱仪的远程控制，这里可以参考我之前的一篇远程Fluke设备的文章，其中原理大致类似。主要是通过pyvisa库实现PC和设备之间的通信，然后再根据设备的官方手册，将其中的SCPI指令按照需求进行封装，从而实现你想要的功能。

由于我这里GPIB线材资源不足，且GPIB的成本较高，所以我这里采用的是Ethernet通信的方式来控制频谱仪。

所以，在开始之前，需要安装NI软件（自行百度），安装完毕后打开NI MAX 添加设备（在此之前需将PC的IP地址改为和频谱仪同一网段，否则无法找到设备）

步骤1：点击展开->右键设备和接口 -> 新建 -> 选择VISA TCP/IP Resources

步骤2： 选择Auto-detect of LAN instrument -> 点击下一步:

步骤3： 选择 Manually specify address information of LAN instrument

步骤4： 点击下一步 -> 输入 Signal Analyzer 的IP地址 -> 点击Validate ->完成：

（若验证失败，请检查PC和Signal Analyzer的IP地址是否在同一网段）。

代码详解

 步骤1： 建立连接

	def _sigAnalyzer_connection(cls, resourceName):# 这里是建立连接的私有函数，主要通过pyvisa库列出所有的资源，判断我的设备是否在已有的资源列表中resourceName = str(resourceName)rm = pyvisa.ResourceManager()resource_list = rm.list_resources(query='?*::INSTR')if resourceName in resource_list:cls.inst = rm.open_resource(resourceName)cls.inst.clear()cls.inst.timeout = 8000return cls.instelse:raise AssertionError("The signal analyzer is not online.")

然后，封装一个打开连接的函数：

@classmethoddef _open_signal_analyzer_connection(cls, ipAddress):ipAddress = str(ipAddress)resourceName = 'TCPIP0::{:s}::inst0::INSTR'.format(ipAddress)connectionID = cls.get_connectionID(resourceName)if connectionID not in cls._connections:cls._connections[connectionID] = cls._sigAnalyzer_connection(resourceName)cls._active_connection = connectionIDprint('Connect to SA successful')return (connectionID, cls._connections[connectionID])

步骤2： 发送SCPI指令

@classmethoddef _send_raw_command(cls, command, connectionID=None):command = str(command)connectionID = cls._check_connectionID(connectionID)if not command:raise ValueError('Command string is empty')_SAKeywords._connections[connectionID].write(command)try:cls._error_query()except:raise IOError('Send command ({:s}) to signal analyzer failed.'.format(command))

步骤3： 根据需求封装函数实现功能

这里以设置频谱仪测试模式的函数为例，首先呢，写一个设备支持的测试模式的枚举类：

class TestMode(enum.Enum):SAN = 0SPEC = 1

然后，在文档中找到设置测试模式的指令为 “CONF:模式名” ，然后直接调用之前的send_raw_comman()方法向设备发送相应的指令即可:

	def set_signal_analyzer_test_mode(self, mode=TestMode.SAN):mode = Utils.to_enum(mode, TestMode)command = 'CONF:{:s}'.format(mode.name)_SAKeywords._send_raw_command(command)print('Set SA to {:s} mode'.format(mode.name))

所有其他功能都是相似的道理，核心在于熟读设备的官方手册，由于这些仪器仪表的手册多为英文，所以有时候英文水平还是相当重要的！！！

项目源码

 我这里控制频谱仪是为了做GSM的带内杂散测试，所以只封装了仪器的部分功能；该频谱仪的功能很全，但是我们常用的只是九牛一毛而已。下面附上源码，只为抛砖引玉，如有错误或者不足之处，还望各位大佬指出，谢谢！

import enum
import timeimport pyvisa
import hashlibfrom utils import Utilsclass SignalError(Exception):passclass SupportedTimeUnit(enum.Enum):KS = 'Ks'S = 's'MS = 'ms'US = 'us'NS = 'ns'class TestMode(enum.Enum):SAN = 0SPEC = 1class SweepMode(enum.Enum):SINGLE = 0CONTINUOUS = 1class RFCouplingMode(enum.Enum):AC = 0DC = 1class TraceType(enum.Enum):WRIT = 0AVER = 1MAXH = 2MINH = 3class TriggerSources(enum.Enum):IMM = 0VID = 1LINE = 2EXT1 = 3EXT2 = 4RFB = 5PRAM = 6TV = 7class TriggerSlope(enum.Enum):POS = 0NEG = 1class MarkerMode(enum.Enum):OFF = 0POSITION = 1DELTA = 2FIXED = 3class BandFunctionType(enum.Enum):OFF = 0NOISE = 1BPOW = 2BDEN = 3class _SAKeywords(object):_connections = {}_active_connection = Nonedef __init__(self):super().__init__()# region Connection Handling@classmethoddef _check_connectionID(cls, connectionID):if connectionID is None:connectionID = cls._active_connectionif connectionID is None:raise SignalError('No signal analyzer connection established')if connectionID not in cls._connections.keys():raise SignalError('No signal analyzer connection with ID {:s} established'.format(connectionID))return connectionID@classmethoddef _open_signal_analyzer_connection(cls, ipAddress):ipAddress = str(ipAddress)resourceName = 'TCPIP0::{:s}::inst0::INSTR'.format(ipAddress)connectionID = cls.get_connectionID(resourceName)if connectionID not in cls._connections:cls._connections[connectionID] = cls._sigAnalyzer_connection(resourceName)cls._active_connection = connectionIDprint('Connect to SA successful')return (connectionID, cls._connections[connectionID])@classmethoddef _close_signal_analyzer_connection(cls, connectionID=None):connectionID = cls._check_connectionID(connectionID)try:cls._connections[connectionID].close()except SignalError as err:print('{!s}'.format(err))del cls._connections[connectionID]cls._active_connection = None if not cls._connections else next(iter(cls._connections.keys()))print('Disconnected with SA')return (cls._active_connection, cls._connections[cls._active_connection] if cls._active_connection else None)@classmethoddef _sigAnalyzer_connection(cls, resourceName):resourceName = str(resourceName)rm = pyvisa.ResourceManager()resource_list = rm.list_resources(query='?*::INSTR')if resourceName in resource_list:cls.inst = rm.open_resource(resourceName)cls.inst.clear()cls.inst.timeout = 8000return cls.instelse:raise AssertionError("The signal analyzer is not online.")@classmethoddef _send_raw_command(cls, command, connectionID=None):command = str(command)connectionID = cls._check_connectionID(connectionID)if not command:raise ValueError('Command string is empty')_SAKeywords._connections[connectionID].write(command)try:cls._error_query()except:raise IOError('Send command ({:s}) to signal analyzer failed.'.format(command))@classmethoddef _send_query_command(cls, command, connectionID=None):command = str(command)connectionID = cls._check_connectionID(connectionID)data_length = len(command)if data_length != 0:try:content = cls._connections[connectionID].query(command)except Exception as err:raise IOError("Query command ({:s}) from signal analyzer failed, {:s}".format(command, err))else:raise AssertionError('This command is not correct!')return content@staticmethoddef get_connectionID(resourceName):return hashlib.sha1(str(resourceName).lower().encode('utf-8')).hexdigest()@classmethoddef _error_query(cls):content = cls._send_query_command(command=':SYST:ERR?')code = content.split(',')[0]if int(code) != 0:raise AssertionError('Already has error:{}'.format(content))def open_signal_analyzer_connection(self, ipAddress):_SAKeywords._open_signal_analyzer_connection(ipAddress)self.signal_analyzer_initialize()def close_signal_analyzer_connection(self):return _SAKeywords._close_signal_analyzer_connection()def _signal_analyzer_reset(self):_SAKeywords._send_raw_command('*CLS')_SAKeywords._send_raw_command('*RST')_SAKeywords._send_raw_command('*OPC')_SAKeywords._send_raw_command(':SYST:PRES')print('Set SA to default setting')def signal_analyzer_initialize(self, reset=True):data = _SAKeywords._send_query_command('*IDN?')if 'N9010B' in data or 'N9030B' in data or 'N9020B' in data or 'N9040B' in data or 'N9000B' in data:if reset is True:self._signal_analyzer_reset()else:passelse:raise AssertionError('Signal analyzer initialize failed: device general information is wrong, {}'.format(data))# endregion# region Test Preparedef set_signal_analyzer_test_mode(self, mode=TestMode.SAN):mode = Utils.to_enum(mode, TestMode)command = 'CONF:{:s}'.format(mode.name)_SAKeywords._send_raw_command(command)print('Set SA to {:s} mode'.format(mode.name))def set_signal_analyzer_sweep_mode(self, mode=SweepMode.CONTINUOUS):mode = Utils.to_enum(mode, SweepMode)command = 'INIT:CONT {:d}'.format(mode.value)_SAKeywords._send_raw_command(command)print('Set SA sweep mode to {:s}'.format(mode.name))def set_signal_analyzer_input_coupling_mode(self, mode=RFCouplingMode.AC):mode = Utils.to_enum(mode, RFCouplingMode)command = 'INP:COUP {:s}'.format(mode.name)_SAKeywords._send_raw_command(command)print('Set SA input coupling mode to {:s}'.format(mode.name))# endregion# region Configurationdef set_signal_analyzer_freq_configuration(self, centerFreq=None, spanFreq=None):""":param centerFreq: e.g 10ghz(GHZ), 10mhz(MHZ), 10khz(KHZ), 10hz(HZ):param spanFreq: e.g 10ghz(GHZ), 10mhz(MHZ), 10khz(KHZ), 10hz(HZ):return:None"""centerFreq = str(centerFreq) if centerFreq is not None else centerFreqspanFreq = str(spanFreq) if spanFreq is not None else spanFreqfreqParams = []print('Set SA frequency to {}'.format(centerFreq))if centerFreq is not None:freqParams.append('CENT {:s}'.format(centerFreq))if spanFreq is not None:freqParams.append('SPAN {:s}'.format(spanFreq))if len(freqParams) >= 1:_SAKeywords._send_raw_command('FREQ:{:s}'.format(';'.join(freqParams)))else:print('No frequency command sent, no parameter was set')def set_signal_analyzer_amplitude_configuration(self, refLevel=None, refLevOffset=None, attenuation=None):refLevel = float(refLevel) if refLevel is not None else refLevelrefLevOffset = float(refLevOffset) if refLevOffset is not None else refLevOffsetattenuation = float(attenuation) if attenuation is not None else attenuationscaleParams = []attParams = []print('Set SA ref level offset to {} dB, attenuation to {}'.format(refLevel, attenuation))if refLevel is not None:scaleParams.append('RLEV {:.2f}'.format(refLevel))if refLevOffset is not None:scaleParams.append('RLEV:OFFS {:.2f}'.format(refLevOffset))if attenuation is not None:attParams.append('ATT {:.2f}'.format(attenuation))if len(scaleParams) >= 1:_SAKeywords._send_raw_command(':DISP:WIND:TRAC:Y:{:s}'.format(';'.join(scaleParams)))else:print('No amplitude y scale command sent, no parameter was set')if len(attParams) >= 1:_SAKeywords._send_raw_command('POW:{:s}'.format(';'.join(attParams)))else:print('No attenuation command sent, no parameter was set')def set_signal_analyzer_bw_configuration(self, resBW=None, videoBW=None):""":param resBW: e.g 10ghz(GHZ), 10mhz(MHZ), 10khz(KHZ), 10hz(HZ):param videoBW: e.g 10ghz(GHZ), 10mhz(MHZ), 10khz(KHZ), 10hz(HZ):return: None"""resBW = str(resBW) if resBW is not None else resBWvideoBW = str(videoBW) if videoBW is not None else videoBWbwParams = []print('Set SA RBW to {}, VBW to {}'.format(resBW, videoBW))if resBW is not None:bwParams.append(':BAND {:s}'.format(resBW))if videoBW is not None:bwParams.append(':BAND:VID {:s}'.format(videoBW))if len(bwParams) >= 1:_SAKeywords._send_raw_command('{:s}'.format(';'.join(bwParams)))else:print('No BW command sent, no parameter was set')# endregion

支持的设备信号

由于我这边设备有限，以上代码只在keysight的N9010B, N9020B， B9030B等仪器上适配过，其他仪器暂未确定是否支持！！！