Привет! В предыдущей статье https://habr.com/ru/articles/1042328/ я рассказывал о математической оценке коэффициента шума (КШ) методом отношения сигнал-шум. Сегодня перейдем от теории к практике - к автоматизации радиоизмерений. Если вы когда-нибудь часами работали с анализатором спектра (АС), записывая показания в блокнот, а потом ещё и переносили их в Excel, — эта статья для вас. Я покажу, как построить гибкую систему управления измерительными приборами на Python, которая не боится разношёрстного парка оборудования в лаборатории.

Боль ручных измерений

Радиоинженеру регулярно приходится проверять характеристики приёмопередающей аппаратуры: коэффициент передачи, КШ, точку 1 дБ компрессии. И если анализатор цепей не спасает (а он точно не спасает при двойном переносе частоты, измерении КШ или когда избирательность радиочастотных фильтров превышает динамический диапазон анализатора цепей), остаётся связка генератор сигналов (ГС) + анализатор спектра (АС).

Ручная работа выглядит так: настроил частоту ГС, настроил центральную частоту АС, поправил RBW, аттенюатор, записал результат измерения, повторил для следующей частоты. На сотню точек уходит пару часов. Знакомо?

Я написал скрипт под свой АКИП-4212/3 — и жизнь наладилась. Но когда АКИП ушел в ремонт, и у меня на руках оказался Keysight N9952A, то пришло понимание, что требуется более основательный подход. Ведь уже на команде установки входного аттенюатора :POW:ATT {value} DB; АС Keysight N9952A выдал ошибку распознавания команды. Пришлось лезть в Programming Guide и дописывать второй драйвер. Тогда я понял: нужна архитектура, которая позволит добавлять команды управления для новых приборы за считанные минуты.

Наивный подход: класс для одного прибора

Когда прибор один, всё просто. Создаём класс с методами, внутри которых прописаны конкретные SCPI-команды:

import time
class SpecAkip4212:

    def __init__(self, host, port):
        # инициализация соединения
        pass
        
    def send(self, cmd, do_receive=0):
        # отправка команды через сокет
        pass
        
    def spec_cent_span_rbw_att(self, freq_cent, freq_span, rbw, att):
        self.send([
            ':INST:MEAS OFF',
            ':POW:GAIN OFF;',
            f':FREQ:CENT {freq_cent} MHz;',
            f':FREQ:SPAN {freq_span} MHz;',
            f':BAND:RES {rbw} MHz;',
            f':POW:ATT {att} DB;',
        ])
        
    def peak_search(self): # поиск максимума
        return self.send([
        ':CALC:MARK1:STAT ON; :CALC:MARK1:MAX; :CALC:MARK1:Y?'], do_receive=1)
    
    def min_peak_search(self): # поиск минимума
		return self.send([
		':CALC:MARK1:STAT ON; :CALC:MARK2:MIN; :CALC:MARK2:X?'], do_receive=1)
		
    def set_cont_peak_on(self): # непрерывное слежение за максимумом
        self.send([':CALC:MARK1:CPE ON'])
        
    def preamp_on(self): # включить предусилитель
		self.send([':POW:GAIN ON;'])
		
    # Установка режима измерения мощности в полосе
    def set_CHP_mode(self, freq_cent, freq_span, rbw, att, intbw):
        self.send([
            ':INST:MEAS CHP',
            ':POW:GAIN OFF;',
            f':FREQ:CENT {freq_cent} MHz;',
            f':FREQ:SPAN {freq_span} MHz;',
            f':POW:ATT {att} DB;',
            f':BAND:RES {rbw} MHz;',
        ])
        time.sleep(1)  # ждём переключения режима
        self.send([f':CHP:BWID:INT {intbw} kHz;'])

Работает отлично. Но только с этим конкретным прибором. Как только появляется второй анализатор — приходится копировать класс, дублировать код и менять команды. А если приборов пять? Начинается хаос.

Почему SCPI-команды разных АС различаются

SCPI-стандарт описывает синтаксис, но не семантику. Каждый производитель добавляет свои подсистемы и меняет названия. Сравним команды для трёх разных анализаторов:

Назначение команды

АКИП-4212/3

АКИП-4214/2

Keysight N9952

Уст. центр. частоты

:FREQ:CENT {v} MHz;

:FREQ:CENT {v} MHz;

:SENS:FREQ:CENT {v} MHz;

Уст. полосы обзора

:FREQ:SPAN {v} MHz;

:FREQ:SPAN {v} MHz;

:SENS:FREQ:SPAN {v} MHz;

Уст. полосы разр.

:BAND:RES {v} MHz;

:BAND:RES {v} MHz;

:SENS:BAND:RES{v} MHz;

Уст. вх. аттен.

:POW:ATT {v} DB;

:POW:ATT {v} DB;

:SENS:POW:RF:ATT {v} DB;

Выкл. реж. изм.

:INST:MEAS OFF;

:INST:MEAS SA;

:SENS:MEAS:CHAN NONE;

Вкл. реж. изм. CHP

:INST:MEAS CHP;

:INST:MEAS CHP;

:SENS:MEAS:CHAN CHP;

Уст. полосы интегр. CHP

:CHP:BWID:INT {v} MHz;

:CHP:BWID:INT {v} MHz;

:SENS:CME:IBW {v} MHZ;

Вкл. предусил.

:POW:GAIN ON;

Предусил. нет

:SENS:POW:RF:GAIN ON;

Выкл. предусил.

:POW:GAIN OFF;

:POW:GAIN OFF;

:SENS:POW:RF:GAIN OFF;

Поиск пика

:CALC:MARK1:MAX; :CALC:MARK1:Y?;

:CALC:MARK1:MAX; :CALC:MARK1:Y?;

:CALC:MARK1:FUNCTION: MAX; :CALC:MARK1:Y?;

Поиск частоты пика

:CALC:MARK1:MAX; :CALC:MARK1:X?;

:CALC:MARK1:MAX; :CALC:MARK1:X?;

:CALC:MARK1:FUNCTION: MAX; :CALC:MARK1:X?;

Непр. слеж. за макс.

:CALC:MARK1:CPE ON

:CALC:MARK1:CPE ON

Функция отсутствует

Чтение мощ-ти CHP

:MEAS:CHP:CHP?

:MEAS:CHP:CHP?

:CALC:MEAS:DATA?

Вкл. усредн. проходов

:AVER:TRAC1:STAT ON

:CME:AVER:ENAB ON

:CME:AVER:ENAB ON

Уст. проход. усредн.

AVER:TRAC:COUN 5

:SENS:AVER:COUN 5

:SENS:AVER:COUN 5

Выкл. усредн. проходов

AVER:TRAC:COUN 1

:SENS:AVER:COUN 1

:SENS:AVER:COUN 1

Вкл. один. свипа

:INIT:CONT OFF; :INIT:IMM

:INIT:CONT OFF; :INIT:IMM

:INIT:CONT OFF; :INIT:IMM

Вкл. непрер. свипа.

:INIT:CONT ON

:INIT:CONT ON

:INIT:CONT ON

Даже внутри парка приборов одного производителя команды могут отличаться! Значит, для каждого прибора нужен свой набор SCPI-команд. Но логика измерений при этом не изменяется. Как избежать дублирования кода?

Архитектура универсального драйвера

Рисунок ниже упрощенно описывает работу универсального драйвера. Идея проста: разделяем поведение (что делать) и данные (какие команды слать).

Рисунок 1 - Архитектура универсального драйвера
Рисунок 1 - Архитектура универсального драйвера

Создаём базовый класс SpecBase, в котором реализованы все высокоуровневые методы: установка частоты, поиск пика, измерение мощности в канале и т.д. А конкретные SCPI-команды выносим в словарь COMMAND_MAP, который переопределяется в каждом дочернем классе.

Чтобы добавить новый прибор, достаточно создать дополнительный класс-наследник и заполнить COMMAND_MAP правильными SCPI-командами в соответствии с Programming Manual измерительного прибора. Логика измерений остается неизменной.

Базовый класс драйвера SpecBase

Рассмотрим реализацию базового класса. Он получает команды из COMMAND_MAP и подставляет значения. Полный код будет в репозитории https://github.com/MAKSAntonio/auto_meas_example_v1, здесь покажу ключевые фрагменты.

import logging  
import socket  
  
class SpecBase(SCPIdevice):  # Базовый класс драйвера  
    # Словарь команд – каждая модель анализатора переопределяет под себя    COMMAND_MAP = {  
        'mode_sa': None,  # установка режима АС  
        'meas_off': None,  # выключение любых режимов измерения  
        'meas_chp': None,  # включение режима измерение мощности в полосе (CHP)  
        'cent_freq': None,  # установка центральной частоты  
        'span': None,  # установка полосы обзора  
        'rbw': None,  # установка полосы разрешения  
        'att': None,  # установка входного аттенюатора  
        'intbw': None,  # установка полосы интегрирования в режиме CHP  
        'preamp_on': None,  # включение предусилителя  
        'preamp_off': None,  # выключение предусилителя АС  
        'peak_y': None,  # поиск пика  
        'peak_x': None,  # поиск частоты пика  
        'min_peak_y': None,  # поиск минимума  
        'cont_peak_on': None,  # непрерывное слежение за максимумом  
        'chp_read': None,  # чтение мощности в полосе в режиме CHP  
        'avg_on': None,  # включить усреднение проходов  
        'avg_count': None,  # число проходов усреднения  
        'avg_off': None,  # выключить усреднение проходов  
        'sweep_single': None,  # включить одиночное свипирование  
        'sweep_cont': None,  # включить непрерывное свипирование  
    }  
  
    def __init__(self, host, port):  
        self.host = host  
        self.port = port  
        
    def send(self, cmds, do_receive=None):  
    """Отправляет команду и ждет ответ, если do_receive"""  
	    if not self.host:  
	        raise Exception(f"{self.title}: не установлен IP")  
	    try:  
	        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:  
	            s.connect((self.host.encode('ASCII'), self.port))  
	            for cmd in cmds:  
	                s.sendall((cmd + "\n").encode('ASCII'))  
	            return (s.recv(1024)) if do_receive else None  
	    except Exception as e:  
	        logging.error(f"({self.host}): ошибка отправки команды.\n{e}")  
	        return False  
  
  
	def wait_for_setting(self, set_cmd, query_cmd, target, attempts=5, delay=0.1): 
	"""Отправляет команду и проверяет (если нужно) ответ по query_cmd.  
	Повторяет attempts раз, чтобы убедиться, что прибор принял настройку."""    
		for attempt in range(attempts):  
	        self.send([set_cmd])  
	        if target is None:  
	            return True  
	        response = self.send([query_cmd], do_receive=1)  
	        if response is None:  
	            time.sleep(delay)  
	            continue  
	        try:  
	            if isinstance(target, str):  
	                current = str(response.decode().strip())  
	                if target in current:  
	                    return True  
	            else:  
	                current = float(response.decode().strip())  
	                if abs(current - target) < 1e-9:  
	                    return True  
	        except ValueError:  
	            pass  
	        time.sleep(delay)  
	    logging.warning(f"Не удалось установить значение {target} за {attempts} попыток. ")  
	    raise RuntimeError(f"Не удалось установить {target} за {attempts} попыток")
  
    def _set_param(self, cmd_template, value, query_template=None, target=None):  
        set_cmd = cmd_template.format(value=value)  
        query_cmd = query_template.format(value=value) if query_template else None  
        self.wait_for_setting(set_cmd, query_cmd, target)  
  
    def set_mode_sa(self):  # Установка режима АС  
        cmd_set, cmd_query, target = self.COMMAND_MAP['mode_sa']  
        if cmd_set is None:  
            print(f'{self.__class__.__name__} в режиме SA по умолчанию')  
            return  
        self.wait_for_setting(set_cmd=cmd_set, query_cmd=cmd_query, target=target)  
  
    def set_meas_off(self):  # выключение любых режимов измерения (например, CHP)  
        cmd_set, cmd_query, target = self.COMMAND_MAP['meas_off']  
        self.wait_for_setting(set_cmd=cmd_set, query_cmd=cmd_query, target=target)  
  
    def set_meas_chp(self):  # включение режима измерение мощности в полосе (CHP)  
        cmd_set, cmd_query, target = self.COMMAND_MAP['meas_chp']  
        self.wait_for_setting(set_cmd=cmd_set, query_cmd=cmd_query, target=target)  
  
    def spec_cent(self, freq_mhz):  # Установка центральной частоты  
        cmd_set, cmd_query, _ = self.COMMAND_MAP['cent_freq']  
        self._set_param(cmd_set, freq_mhz, cmd_query, int(freq_mhz * 1e6))  
  
    def spec_span(self, freq_span_mhz):  # Установка полосы обзора  
        cmd_set, cmd_query, _ = self.COMMAND_MAP['span']  
        self._set_param(cmd_set, freq_span_mhz, cmd_query, None)  
  
    def spec_rbw(self, rbw_mhz):  # Установка полосы разрешения  
        cmd_set, cmd_query, _ = self.COMMAND_MAP['rbw']  
        self._set_param(cmd_set, rbw_mhz, cmd_query, None)  
  
    def spec_att(self, att_db):  # Установка входного аттенюатора  
        cmd_set, cmd_query, _ = self.COMMAND_MAP['att']  
        self._set_param(cmd_set, att_db, cmd_query, None)  
  
    # Установка полосы интегрирования в режиме CHP  
    def spec_intbw(self, intbw_mhz):  
        cmd_set, cmd_query, _ = self.COMMAND_MAP['intbw']  
        self._set_param(cmd_set, intbw_mhz, cmd_query, None)  
  
    def preamp_on(self):  # Включение предусилителя  
        cmd_set, cmd_query, target = self.COMMAND_MAP['preamp_on']  
        if cmd_set is None:  
            raise RuntimeError(f'{self.__class__.__name__} не поддерживает предусилитель')  
        self.wait_for_setting(set_cmd=cmd_set, query_cmd=cmd_query, target=target)  
  
    def preamp_off(self):  # Выключение предусилителя  
        cmd_set, cmd_query, target = self.COMMAND_MAP['preamp_off']  
        if cmd_set is None:  
            return  # Нет предусилителя – просто выходим  
        self.wait_for_setting(set_cmd=cmd_set, query_cmd=cmd_query, target=target)  
  
    def prepare_single_sweep(self):  # включить одиночное свипирование  
        self.send([self.COMMAND_MAP['sweep_single'], '*OPC?'], do_receive=1)  
  
    def set_cont_sweep(self):  # включить непрерывное свипирование  
        self.send([self.COMMAND_MAP['sweep_cont']])  
  
    def peak_search(self):  # Поиск пика с запросом  
        self.prepare_single_sweep()  
        cmd = self.COMMAND_MAP['peak_y']  
        result = self.send([cmd], do_receive=1)  
        return float(result.decode()) if result else None  
  
    def peak_search_freq(self):  # Поиск частоты пика  
        self.prepare_single_sweep()  
        cmd = self.COMMAND_MAP['peak_x']  
        result = self.send([cmd], do_receive=1)  
        return float(result.decode()) if result else None  
  
    def min_search(self):  # Поиск следующего пика  
        self.prepare_single_sweep()  
        cmd = self.COMMAND_MAP['min_peak_y']  
        result = self.send([cmd], do_receive=1)  
        return float(result.decode()) if result else None  
  
    def set_cont_peak_on(self):  # Установка следящего peak_search  
        cmd = self.COMMAND_MAP['cont_peak_on']  
        if cmd == None:  
            print(f'{self.__class__.__name__} не поддерживает функцию следящего поиска максимума')  
            return  
        self.send([cmd], do_receive=0)  
  
    def CHP_meas(self):  # Измерение мощности в полосе intbw в режиме CHP  
        avg_on_cmd = self.COMMAND_MAP['avg_on']  
        avg_cnt_cmd = self.COMMAND_MAP['avg_count']  # усреднение 5 раз  
        self.send([avg_on_cmd, avg_cnt_cmd])
        self.prepare_single_sweep()
        read_cmd = self.COMMAND_MAP['chp_read']  
        result = self.send([read_cmd], do_receive=1)  
        avg_off_cmd = self.COMMAND_MAP['avg_off']  
        self.send([avg_off_cmd])  
        if result:  
            return float(result.decode().split(',')[0])  
        return None  
  
    def spec_cent_span_rbw_att(self, freq_cent, freq_span, rbw, att):  
        self.set_meas_off()  
        self.preamp_off()  
        self.spec_cent(freq_cent)  
        self.spec_span(freq_span)  
        self.spec_rbw(rbw)  
        self.spec_att(att)  
  
    def set_CHP_mode(self, freq_cent, freq_span, rbw, att, intbw):  
        self.set_meas_chp()  
        self.preamp_on()  
        self.spec_cent(freq_cent)  
        self.spec_intbw(intbw)  
        self.spec_span(freq_span)  
        self.spec_rbw(rbw)  
        self.spec_att(att)

Как видите, все методы обращаются к self.COMMAND_MAP[...]. Именно там хранятся конкретные команды. Теперь посмотрим, как заполняются словари разных приборов.

Конкретные драйверы АС: заполняем словари

АКИП-4212/3:

class Akip4212Driver(SpecBase):  
    COMMAND_MAP = {  
        'mode_sa': (None, None, None),  
        'meas_off': (':INST:MEAS OFF;', ':INST:MEAS?;', 'OFF'),  
        'meas_chp': (':INST:MEAS CHP;', ':INST:MEAS?;', 'CHP'),  
        'cent_freq': (':FREQ:CENT {value} MHz;', ':FREQ:CENT?;', None),  
        'span': (':FREQ:SPAN {value} MHz;', ':FREQ:SPAN?;', None),  
        'rbw': (':BAND:RES {value} MHz;', ':BAND:RES?;', None),  
        'att': (':POW:ATT {value} DB;', ':POW:ATT?;', None),  
        'intbw': (':CHP:BWID:INT {value} MHz;', ':CHP:BWID:INT?;', None),  
        'preamp_on': (':POW:GAIN ON;', ':POW:GAIN?;', 1),  
        'preamp_off': (':POW:GAIN OFF;', ':POW:GAIN?;', 0),  
        'peak_y': ':CALC:MARK1:MAX; :CALC:MARK1:Y?',  
        'peak_x': ':CALC:MARK1:MAX; :CALC:MARK1:X?',  
        'min_peak_y': ':CALC:MARK2:MIN; :CALC:MARK2:Y?',  
        'cont_peak_on': ':CALC:MARK1:CPEak ON',  
        'chp_read': ':MEAS:CHP:CHP?',  
        'avg_on': ':AVER:TRAC1:STAT ON',  
        'avg_count': 'AVER:TRAC:COUNt 5',  
        'avg_off': 'AVER:TRAC:COUNt 1',  
        'sweep_single': ':INIT:CONT OFF; :INIT:IMM',  
        'sweep_cont': ':INIT:CONT ON',  
    }

Keysight N9952A:

class KeysightN9952ADriver(SpecBase):  
    COMMAND_MAP = {  
        'mode_sa': ('INST "SA";', 'INST?;', 'SA'),  
        'meas_off': (':SENS:MEAS:CHAN NONE;', ':SENS:MEAS:CHAN?;', 'NONE'),  
        'meas_chp': (':SENSe:MEAS:CHAN CHP;', ':SENSe:MEAS:CHAN?;', 'CHP'),  
        'cent_freq': (':SENS:FREQ:CENTer {value} MHZ;', ':SENS:FREQ:CENT?;',None), 
        'span': (':SENS:FREQ:SPAN {value} MHZ;', ':SENSe:FREQ:SPAN?;', None),  
        'rbw': (':SENS:BAND:RES {value} MHZ;', ':SENS:BAND:RES?;', None),  
        'att': (':SENS:POW:RF:ATT {value} DB;', ':SENS:POW:RF:ATT?;', None),  
        'intbw': (':SENS:CME:IBW {value} MHZ', ':SENSe:CME:IBW?;', None),  
        'preamp_on': (':SENS:POW:RF:GAIN ON;', ':SENS:POW:RF:GAIN?;', 1),  
        'preamp_off': (':SENS:POW:RF:GAIN OFF;', ':SENS:POW:RF:GAIN?;', 0),  
        'peak_y': ':CALC:MARK1:FUNCTION:MAX; :CALC:MARK1:Y?',  
        'peak_x': ':CALC:MARKer1:FUNCTION:MAX; :CALC:MARKer1:X?',  
        'min_peak_y': ':CALC:MARK2:FUNC:MIN; :CALC:MARK2:Y?',  
        'cont_peak_on': None,  
        'chp_read': ':CALC:MEAS:DATA?',  
        'avg_on': ':CME:AVER:ENAB ON',  
        'avg_count': ':SENS:AVER:COUN 5',  
        'avg_off': ':SENS:AVER:COUN 1',  
        'sweep_single': ':INIT:CONT OFF; :INIT:IMM',  
        'sweep_cont': ':INIT:CONT ON',  
    }  
    def spec_att(self, att_db):  
        """Переопределение метода: ограничение аттенюатора для N9952A"""  
        if att_db > 30:  
            att_db = 30  
            logging.error('Макс. значение ATT для Keysight N9952A равно 30 дБ')  
        super().spec_att(att_db)

Для Keysight N9952A возникла необходимость переопределить метод установки ослабления аттенюатора, так как максимальное допустимое значение не превышает 30 дБ.

Выбор драйвера по *IDN?

Осталось научиться автоматически определять, какой драйвер использовать. При подключении к прибору отправляем *IDN? и по ответу выбираем класс.

DRIVER_CLASSES = [  
    Akip4212Driver,  
    KeysightN9952ADriver,  
    Akip4214Driver  
]  
IDN_MAP = {  
    'Akip': Akip4212Driver,  
    'Keysight': KeysightN9952ADriver,  
    'Vendor YSA001 13.6G': Akip4214Driver, # пример нестандартного ответа  
}  
def select_spec_driver(host, port=5025):  
    try:  
        socket.create_connection((host, port), timeout=1)  
        temp_driver = SCPIdevice(host, port)  
        idn_response = temp_driver.send(['*IDN?'], do_receive=1).decode().split(',')[0]  
        if idn_response == 'Vendor':  
            idn_response = temp_driver.send(['*IDN?'], do_receive=1).decode().split(',')[0:2]  
            idn_response = idn_response[0] + ' ' + idn_response[1]  
        print(f"Получен *IDN?: {idn_response}")  
        for keyword, driver_class in IDN_MAP.items():  
            if keyword.lower() in idn_response.lower():  
                print(f"Обнаружен прибор: {keyword}")  
                inst = driver_class(host, port)  
                inst.set_mode_sa()  
                return inst  
        print(f"Неизвестный прибор: {idn_response}. Используем первый драйвер.")  
        inst = DRIVER_CLASSES[0](host, port)  
        inst.set_mode_sa()  
        return inst  
    except (socket.timeout, socket.error):  
        print(f"Не удалось подключиться к {host}:{port}")  
        return False

Приведенный участок кода:

  • проверяет доступность прибора по TCP

  • читает идентификатор

  • сопоставляет с ключевыми словами в IDN_MAP

  • при необходимости извлекает дополнительную информацию из ответа

  • создает экземпляр нужного драйвера и инициализирует режим

Исследование быстродействия: почему time.sleep не решает вопрос перестройки настроек АС

Рассмотрим важную деталь, неочевидно заложенную в приведенный выше код.
Например, пользователю нужно сменить центральную частоту на АС и на новой установленной частоте выполнить поиск пика. Если выполнить последовательную запись по SCPI команд :FREQ:CENT 1000 MHz; :CALC:MARK1:MAX; :CALC:MARK1:Y;?, то актуальное значение поиска пика получено не будет. АС требуется перестроить внутренний гетеродин на другую частоту, выполнить хотя бы один один проход (sweep/свип) по указанному частотному диапазону (SPAN) и только после этого найти максимум среди всех значений. Для выполнения перечисленных действия требуется время. Эмпирически было установлено, что при фиксированном отношении RBW:SPAN = 1:100 для большинства современных АС достаточно ввести задержку в 200 мс. То есть последовательность команд должна быть следующая:

  • :FREQ:CENT 1000 MHz;

  • time.sleep(0.2)

  • :CALC:MARK1:MAX;

  • :CALC:MARK1:Y?;

Такой подход оказался ограниченным не только фиксированным соотношением RBW:SPAN=1:100, но и тем, что некоторые АС при непрерывной работе на несколько тысяч измерений на программном уровне могут подвиснуть на секунду и более. Это приводит к тому, что АС может зафиксировать целый ряд ложных показаний. На практике вероятность ошибки при таких измерениях достигала 0,4%: на 1000 измерений приходилось до 40 ошибок измерения.
Универсальное решение - использовать механизм завершения операций, встроенный в SCPI: команду OPC?. Она возвращает 1, когда все предыдущие команды выполнены. А еще лучше перевести АС в режим одиночного свипа (INIT:CONT OFF; INIT:IMM) и ждать завершения OPC?. В приведенном базовом классе SpecBase это реализовано в методе prepare_single_sweep() :

def prepare_single_sweep(self): # включить одиночное свипирование
	self.send([self.COMMAND_MAP['sweep_single'], '*OPC?'], do_receive=1)  

Здесь команда *OPC? блокирует выполнение до тех пор, пока прибор не закончит текущий свип. Теперь нет никаких магических задержек — только аппаратная синхронизация. Методы peak_search()peak_search_freq() и CHP_meas() вызывают prepare_single_sweep() перед чтением данных, поэтому результат всегда актуален.

Практический пример использования

Теперь покажем, как применять написанный драйвер в реальной задаче - автоматизированной оценке коэффициента шума (КШ) приемного аналогового тракта (RX). Математика методологии измерения подробно разобрана в статье https://habr.com/ru/articles/1042328/, поэтому сосредоточимся на программной реализации.
Для оценки КШ используется метод отношения сигнал-шум. Алгоритм:

  1. ГС подключен ко входу RX, формирование сигнала с ГС выключено - эквивалентно подключенной согласованной нагрузке (СН) на входе RX.

  2. Измеряем мощность шума на выходе RX в заданной полосе частот.

  3. На вход RX с ГС подаем гармонический сигнал известного уровня на частоте приема.

  4. Снова измеряем мощность на выходе RX в заданной полосе частот.

  5. Вычисляем КШ по формула из прошлой статьи. Приведенный ниже скрипт автоматически проходит по списку частот freq_array, на каждой частоте производит измерение пунктов 2) и 4), вычисляет КШ и сохраняет результат в текстовый файл:

import numpy as np  
import pickle  
import math  
import matplotlib.pyplot as plt  
import os  
from scipy.interpolate import interp1d  
from drivers.spec_drivers import select_spec_driver  
from drivers.gen_drivers import select_gen_driver  
from drivers.board_drivers import select_board_driver  
  
class meas:  
    def __init__(self, b, g, s):  
        self.b = b  
        self.g = g  
        self.s = s  
  
    def measONE_NF(self, f, intbw, level, name_cableIN_pickle):  
        SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))  
        BASE_DIR = os.path.dirname(SCRIPT_DIR)  
        path_calibIN = os.path.join(BASE_DIR, 'calib_cables', name_cableIN_pickle)  
        # path_calibIN = 'calib_cables\\' + name_cableIN_pickle  
        with open(path_calibIN, 'rb') as file_calib:  # загрузка ослабления  
            loss_cableIN = pickle.load(file_calib)  # входного кабеля  
        k = 1.38 * 10 ** (-23)  # Дж/К - Постоянная Больцмана  
        N0 = k * 300  # ДЖ = ВТ*с - СПМ теплового шума  
        self.g.gen_OFF()  
        self.s.spec_cent(f)  
        q1_db = self.s.CHP_meas()  # измер. мощ-ти собст. шумов на выходе RX  
        q1 = (10 ** (q1_db / 10))  # перевод дБмВт в мВт  
        self.g.gen_freq_level(f, level)  
        self.g.gen_ON()  
        q2_db = self.s.CHP_meas()  # измер. мощ-ти собст. шумов + сигнал на вых RX  
        q2 = (10 ** (q2_db / 10))  # перевод дБмВт в мВт  
        We = 10 ** ((level - loss_cableIN(f) - 30) / 10) / (intbw * 1e6)  # СПМ сигн. на входе RX  
        NF = 10 * math.log10(q1 / (q2 - q1) * We / N0)  # Оценка КШ  
        self.g.gen_OFF()  
        return np.round(NF, 1)  
  
    def meas_NF(self, farray, name_pickle):  
        SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))  
        BASE_DIR = os.path.dirname(SCRIPT_DIR)  
  
        span = 0.042  # MHz  
        rbw = 0.0001  # MHz: Соотношение span/rbw лучше делать 100  
        att = 0  # dB  
        intbw = 0.042  # MHz  
        level = -60  # дБмВт на выходе ГС  
        self.s.set_CHP_mode(farray[0], span, rbw, att, intbw)  
        NF_full_table = []  
        self.b.power_ON()  
        for f in farray:  
            NF = m.measONE_NF(f, intbw, level, name_pickle)  
            NF_full_table.append([f, NF])  
            print(f'freq = {f} -> NF = {round(NF, 1)} dB')  
        self.b.power_OFF()  
        self.s.set_cont_sweep()  
        self.s.set_meas_off()  
        name_dir = os.path.join(BASE_DIR, 'results')  
        if os.path.exists(name_dir) and os.path.isdir(name_dir):  
            print(f"Директория '{name_dir}' уже существует.")  
        else:  
            os.makedirs(name_dir)  # создаёт все недостающие родительские папки  
            print(f"Директория '{name_dir}' создана.")  
  
        file_path = f'{name_dir}\\nf_meas.txt'  
        with open(file_path, 'w') as file:  
            file.write('\t|'.join(map(str, ['f,MHz', 'NF,dB\n'])))  
            for line in NF_full_table:  
                file.write('\t|\t'.join(map(str, line)) + '\n')  
  
        f_d = [item[0] for item in NF_full_table]  
        NF_d = [item[1] for item in NF_full_table]  
  
        print(f'\nИзмерение КШ успешно выполнено!\n')  
  
        plt.plot(f_d, NF_d, 'bo', label='Измеренные значения')  # синие точки  
        plt.grid()  
        plt.title('Измеренный КШ')  
        plt.xlabel('Частота, МГц')  
        plt.ylabel('КШ, дБ')  
        plt.show()  
  
  
if __name__ == '__main__':
	b = select_board_driver()
	g = select_gen_driver('192.168.1.23')
	s = select_spec_driver('192.168.1.24')
	
	if (b != False)and(g != False)and(s != False):  
	    m = meas(b, g, s)
	else:  
	    print('Обеспечьте корректное подключение стенда')
	
	m.meas_NF(farray=np.arange(950, 2151, 10), name_cableIN_pickle='loss_cableIN.pickle')

Пояснения к коду:

  1. Класс meas инкапсулирует логику измерения и использует отдельные драйвера управления платой, ГС и АС.

  2. Метод measONE_NF выполняет измерение КШ на одной частоте из массива farray. Измерение мощности собственных шумов RX и мощности собственных шумов RX + сигнал через усреднение измеряемой мощности (5 проходов для усреднения, см. метод CHP_meas)

  3. Ослабление входного РЧ кабеля загружается из pickle-файла. Подразумевается, что этот файл был получен на предыдущем этапе: измерение входного РЧ кабеля. Полная версия кода с методом meas_cable приведена в репозитории https://github.com/MAKSAntonio/auto_meas_example_v1.

  4. Сохранение результатов ведётся в текстовый файл nf_meas.txt.

Время измерения одной точки занимает не более 2-3 секунд. Всего 121 измерение, следовательно, требуется около 5 минут для автоматизированного измерения. Ручное измерение того же объема заняло бы часы и было бы подвержено ошибкам оператора.

Заключение

В результате анализа особенностей управления разными АС был показан процесс построения универсального SCPI-драйвера, который легко масштабирует. Предложенная архитектура на основе базового класса SpecBase и словарей команд COMMAND_MAP в классах-наследниках позволяет:

  • Единообразно управлять разными моделями АС, не меняя логику измерений.

  • Добавлять новый прибор за несколько минут: создать класс-наследник, заполнить словарь команд.

  • Избежать дублирования кода и связанных с ним ошибок.

Практический пример оценки КШ продемонстрировал, как можно автоматизировать радиоизмерения. Автоматизация радиоизмерений не ограничивается измерением коэффициента передачи или оценкой КШ. Большой объем автоматизированных радиоизмерений завязан на калибровке радиоаппаратуры: линеаризации коэффициента передачи, оценке входной мощности на входе приемника по известной мощности на выходе АЦП, регулировке напряжений смещения оконечного усилителя для максимизации КПД и других решениях. Для серии таких изделий подобные автоматизации способны экономить месяцы работы и финансовые расходы.

Полный код драйверов и с примерами радиоизмерений доступны в репозитории https://github.com/MAKSAntonio/auto_meas_example_v1. Надеюсь, что предложенный подход сэкономит вам время и поможет сосредоточиться на высокоуровневых инженерных задачах, а не на борьбе с унификацией управления измерительными приборами.

Буду рад обсудить в комментариях ваши сценарии автоматизации и особенности работы. Если у вас есть собственные наработки по универсальным драйверам - делитесь!