Open RF Prototyping

>>> from rfblocks import hmc833, HMC833Controller, create_serial, write_cmd
>>> ser_dev = '/dev/cu.usbmodem14201'
>>> ser = create_serial(ser_dev)
>>> plo = hmc833(sen='D0', ld_sdo='C4', ref='D1', fref=100.0, refdiv=1, refsrc=hmc833.ReferenceSource.EXTERNAL)
>>> ctl = HMC833Controller('plo1', plo)
>>> ctl.initialize(ser)
True
>>> ctl.freq = 500
>>> ctl.configure_freq(ser)
True
>>> ctl.refsrc = hmc833.ReferenceSource.INTERNAL
>>> ctl.ref_freq = 50.0
>>> ctl.ref_div = 1
>>> ctl.configure_refsrc(ser)
>>> ctl.refsrc = hmc833.ReferenceSource.EXTERNAL
>>> ctl.ref_freq = 160.0
>>> ctl.ref_div = 2
>>> ctl.configure_refsrc(ser)
>>> ctl.ref_freq = 70.0
>>> ctl.ref_div = 1
>>> ctl.configure_refsrc(ser)

PLO_SEN = 'D1'
PLO_LD_SDO = 'C4'

serial_device = '/dev/cu.usbmodem14201'

#
# Generated from hmc833.org
#
import sys
from time import sleep
from pathlib import Path
import json
import numpy as np
from scipy import (
    interpolate
)
import rpyc
from rfblocks import hmc833, HMC833Controller, create_serial
from tam import (
    InstrumentMgr, InstrumentInitializeException,
    UnknownInstrumentModelException, HP8560A_GPIBID,
    SMHU58_GPIBID
)

<<serial-port>>
<<pin-assignment>>
do_stage_1 = False
do_stage_2 = False
do_stage_3 = True
ref_level = 20.0

sa_cal_file = 'hmc833-pwr-sa-cal.json'
pwrmeter_cal_file = 'PowerMeter/app/cal-files/24dB-atten-48-0007-Det3-100-6000MHz.json'
stage1_measurements_file = 'hmc833-output-pwr-stage-1.json'
stage2_measurements_file = 'hmc833-output-pwr-stage-2.json'
stage3_measurements_file = 'hmc833-output-pwr-stage-3.json'
loss_fn = None

def insertion_loss(freq):
    return float(loss_fn(freq))

def measure_sigpwr():
    sig_pwr = {}
    for f in freq_list:
        print(f'{f=}:', end='')
        plo_ctl.freq = f
        is_locked = plo_ctl.configure_freq(ser)
        if is_locked:
            pwr, _ = sa.measure_pwr(f)
            sleep(0.5)
            sig_pwr['{:.1f}'.format(f)] = pwr + insertion_loss(f)
            print(f', {pwr+insertion_loss(f)}')
        else:
            print(', PLO lock failed!')
    return sig_pwr

def measure_integrated_pwr():
    sig_pwr = {}
    for f in freq_list:
        print(f'{f=}:', end='')
        plo_ctl.freq = f
        is_locked = plo_ctl.configure_freq(ser)
        if is_locked:
            chan_ctl.freq = f
            pwrmeter.root.measure(pwr_meter_chan)
            sig_pwr['{:.1f}'.format(f)] = float(chan_ctl.pwr)
            print(f', {chan_ctl.pwr}')
        else:
            print(', PLO lock failed!')
    return sig_pwr

def measure_equalized_power():
    sig_pwr = {}
    for f in freq_list:
        print(f'{f=}:', end='')
        if f < 1500.0:
            plo_ctl.divider_gain = hmc833.DividerGain.MAXGAIN
        elif f >= 3000.0:
            plo_ctl.buffer_gain = hmc833.OutputBufferGain.MAXGAIN_MINUS_3DB
        elif f >= 4400.0:
            plo_ctl.buffer_gain = hmc833.OutputBufferGain.MAXGAIN
        plo_ctl.freq = f
        is_locked = plo_ctl.configure(ser)
        if is_locked:
            chan_ctl.freq = f
            pwrmeter.root.measure(pwr_meter_chan)
            sig_pwr['{:.1f}'.format(f)] = float(chan_ctl.pwr)
            print(f', {chan_ctl.pwr}')
        else:
            print(', PLO lock failed!')
    return sig_pwr

def calibrate(srcpwr, start_freq, stop_freq, step_freq):
    loss = {}
    sg.output = True
    for f in np.arange(start_freq, stop_freq, step_freq):
        # Note that the property assignment here may actually
        # be a rpyc remote operation depending on the selected
        # calibration device.  Since ~f~ (and possibly ~srcpwr~)
        # may be numpy array scalars it is necessary to explicitly
        # cast these to standard Python float values.  If this is
        # not done the remote rpyc process will almost certainly
        # generate strange numpy related exceptions.
        sig_freq = float(f)
        sg.freq = sig_freq
        sg.level = float(srcpwr)
        sleep(0.5)
        meas_pwr, _ = sa.measure_pwr(sig_freq)
        loss[sig_freq] = srcpwr - meas_pwr
        sleep(0.5)
    sg.output = False
    freqs = np.array([f for f in loss.keys()])
    loss_arr = np.array([v for v in loss.values()])
    loss_interp_fn = interpolate.interp1d(
        freqs, loss_arr, kind='cubic')
    with open(sa_cal_file, 'w') as fd:
        json.dump(loss, fd)
    return loss_interp_fn

# Test HMC833 PLO initialization
ser = create_serial(serial_device)
plo = hmc833(sen=PLO_SEN, ld_sdo=PLO_LD_SDO)
plo_ctl = HMC833Controller('plo1', plo)
plo_ctl.initialize(ser)

if do_stage_1 is True:
    print()
    print("""---------- Stage 1 ----------\n""")
    print("""Output power measurements from 0.1 to 2.85 GHz.""")
    print("""These measurements use the HP8560A spectrum analyzer.""")
    input("Press a key when ready... ")
    print()

    # Spectrum analyzer initialization
    instrument_mgr = InstrumentMgr()
    try:
        sa = instrument_mgr.open_instrument('HP8560A', HP8560A_GPIBID)
    except InstrumentInitializeException as iie:
        print(iie.message)
        sys.exit(-2)
    except UnknownInstrumentModelException as ume:
        print(ume.message)
        sys.exit(-3)

    sa.clear()
    sa.vavg = 2
    sa.fspan = 0.1
    sa.ref_level = ref_level

    use_existing_sa_cal = False
    cal_path = Path('./'+sa_cal_file)
    if cal_path.is_file():
        print("""There is an existing test cable calibration file: {}.
Would you like to use this?""".format(sa_cal_file))
        ch = input("Use existing cable cal. file? (Y/N): ")
        if ch[0] in ['y', 'Y']:
            use_existing_sa_cal = True

    if not use_existing_sa_cal:
        try:
            sg = instrument_mgr.open_instrument('SMHU58', SMHU58_GPIBID)
        except InstrumentInitializeException as iie:
            print(iie.message)
            sys.exit(-2)
        except UnknownInstrumentModelException as ume:
            print(ume.message)
            sys.exit(-3)

        print("""Calibration of the test cable must now be carried out.
Please connect the test cable between the RF output
of the SMHU58 signal generator and the HP8560A spectrum analyzer.""")
        input("Press a key when ready... ")

        loss_fn = calibrate(0.0, 100.0, 2850.0, 100.0)

        print("Calibration complete.")
    else:
        with open(sa_cal_file) as fd:
            loss = json.load(fd)
        freqs = np.array([float(f) for f in loss.keys()])
        loss_arr = np.array([v for v in loss.values()])
        loss_fn = interpolate.interp1d(
            freqs, loss_arr, kind='cubic')
        print("Cable calibration data loaded from '{}'".format(sa_cal_file))

    print("""Please connect the test cable between the output
of the RF signal generator under test and the input of
the HP8560A spectrum analyzer.""")
    input("Press a key when ready... ")

    freq_list = [float(f) for f in np.linspace(100.0, 2800.0, 40).round()]

    plo_ctl.divider_gain = hmc833.DividerGain.MAXGAIN
    plo_ctl.vco_mute = False

    print("Gain setting: MAXGAIN_MINUS_9DB...")
    plo_ctl.buffer_gain = hmc833.OutputBufferGain.MAXGAIN_MINUS_9DB
    plo_ctl.configure(ser)
    sa.clear()
    sleep(2.0)
    sa.vavg = 2
    sa.fspan = 0.1
    sa.ref_level = ref_level
    max_minus_9dB = measure_sigpwr()

    print("Gain setting: MAXGAIN_MINUS_6DB...")
    plo_ctl.buffer_gain = hmc833.OutputBufferGain.MAXGAIN_MINUS_6DB
    plo_ctl.configure(ser)
    sa.clear()
    sa.vavg = 2
    sa.fspan = 0.1
    sa.ref_level = ref_level
    max_minus_6dB = measure_sigpwr()

    print("Gain setting: MAXGAIN_MINUS_3DB...")
    plo_ctl.buffer_gain = hmc833.OutputBufferGain.MAXGAIN_MINUS_3DB
    plo_ctl.configure(ser)
    sa.clear()
    sa.vavg = 2
    sa.fspan = 0.1
    sa.ref_level = ref_level
    max_minus_3dB = measure_sigpwr()

    print("Gain setting: MAXGAIN...")
    plo_ctl.buffer_gain = hmc833.OutputBufferGain.MAXGAIN
    plo_ctl.configure(ser)
    sa.clear()
    sa.vavg = 2
    sa.fspan = 0.1
    sa.ref_level = ref_level
    maxgain = measure_sigpwr()

    stage_1 = {'max_minus_9dB': max_minus_9dB,
               'max_minus_6dB': max_minus_6dB,
               'max_minus_3dB': max_minus_3dB,
               'maxgain': maxgain}

    # Stash the measurements obtained so far
    with open(stage1_measurements_file, 'w') as fd:
        json.dump(stage_1, fd)

    plo_ctl.vco_mute = True
    plo_ctl.configure(ser)

if do_stage_2 is True:
    print()
    print("""---------- Stage 2 ----------\n""")
    print("""Output power measurements from 0.1 to 6 GHz.""")
    print("""Please attach the power meter 0.1-6GHz measurement head to
the HMC833 PLO module RF output.  Ensure that the measurement head is
connected as channel 0 on the power meter base unit and that the
pwrmeter app is running.""")
    input("Press a key when ready... ")
    print()

    # Power meter initialization.
    pwrmeter = rpyc.connect('127.0.0.1', 18863)
    pwrmeter.root.initialize()
    chan_controllers = pwrmeter.root.detectors
    pwr_meter_chan = 0
    chan_ctl = chan_controllers[pwr_meter_chan]
    chan_controllers[1].enabled = False
    chan_ctl.load_caldata(pwrmeter_cal_file)
    chan_ctl.apply_correction = True

    freq_list = [float(f) for f in np.linspace(100.0, 6000.0, 50).round()]

    plo_ctl.divider_gain = int(hmc833.DividerGain.MAXGAIN)
    plo_ctl.vco_mute = False

    print("Gain setting: MAXGAIN_MINUS_9DB...")
    plo_ctl.buffer_gain = hmc833.OutputBufferGain.MAXGAIN_MINUS_9DB
    plo_ctl.configure(ser)
    max_minus_9dB = measure_integrated_pwr()

    print("Gain setting: MAXGAIN_MINUS_6DB...")
    plo_ctl.buffer_gain = hmc833.OutputBufferGain.MAXGAIN_MINUS_6DB
    plo_ctl.configure(ser)
    max_minus_6dB = measure_integrated_pwr()

    print("Gain setting: MAXGAIN_MINUS_3DB...")
    plo_ctl.buffer_gain = hmc833.OutputBufferGain.MAXGAIN_MINUS_3DB
    plo_ctl.configure(ser)
    max_minus_3dB = measure_integrated_pwr()

    print("Gain setting: MAXGAIN...")
    plo_ctl.buffer_gain = hmc833.OutputBufferGain.MAXGAIN
    plo_ctl.configure(ser)
    maxgain = measure_integrated_pwr()

    stage_2 = {'max_minus_9dB': max_minus_9dB,
               'max_minus_6dB': max_minus_6dB,
               'max_minus_3dB': max_minus_3dB,
               'maxgain': maxgain}

    # Stash the measurements obtained so far
    with open(stage2_measurements_file, 'w') as fd:
        json.dump(stage_2, fd)

    plo_ctl.vco_mute = True
    plo_ctl.configure(ser)

if do_stage_3 is True:
    print()
    print("""---------- Stage 3 ----------\n""")
    print("""Output power measurements from 0.1 to 6 GHz.""")
    print("""Please attach the power meter 0.1-6GHz measurement head to
the HMC833 PLO module RF output.  Ensure that the measurement head is
connected as channel 0 on the power meter base unit and that the
pwrmeter app is running.""")
    input("Press a key when ready... ")
    print()

    # Power meter initialization.
    pwrmeter = rpyc.connect('127.0.0.1', 18863)
    pwrmeter.root.initialize()
    chan_controllers = pwrmeter.root.detectors
    pwr_meter_chan = 0
    chan_ctl = chan_controllers[pwr_meter_chan]
    chan_controllers[1].enabled = False
    chan_ctl.load_caldata(pwrmeter_cal_file)
    chan_ctl.apply_correction = True

    freq_list = [float(f) for f in np.linspace(100.0, 6000.0, 50).round()]

    plo_ctl.divider_gain = int(hmc833.DividerGain.MAXGAIN)
    plo_ctl.vco_mute = False

    equalized_pwr = measure_equalized_power()

    stage_3 = {'equalized_pwr': equalized_pwr}

    # Stash the measurements obtained so far
    with open(stage3_measurements_file, 'w') as fd:
        json.dump(stage_3, fd)

    plo_ctl.vco_mute = True
    plo_ctl.configure(ser)