Open RF Prototyping

def initialize(self) -> None:
    """Initialize the service app and associated hardware modules.

    >>> rfgen = rpyc.connect('127.0.0.1', 18864)
    >>> rfgen.root.initialize()
    """

def initialize_detector(self, chan_id: str) -> bool:
    """Initialize the power detector connected to the specified channel.

    :param chan_id: A signal generator channel id.
        This will be one of "Chan 1" or "Chan 2".
    :type chan_id: str

    :return: True if there is a detector connected, False if no
        detector connected.
    """

@property
def channels(self) -> Dict[str, PLOChan]:
    """Return a dictionary of the signal generator channel objects.

    :return: A dictionary of ``PLOChan`` instances keyed using the
    signal generator channel id  ("Chan 1" and "Chan 2").

    >>> rfgen = rpyc.connect('127.0.0.1', 18864)
    >>> rfgen.root.initialize()
    >>> chans = rfgen.root.channels

    >>> chan_id = 'Chan 1'
    >>> chans[chan_id].plo_ctl.freq = 1500.0
    >>> rfgen.root.configure_plo_freq(chan_id)

    """

def configure_plo_freq(self, chan_id: str) -> None:
    """Update the PLO module output frequency for the specified channel.

    :param chan_id: A signal generator channel id.
        This will be one of "Chan 1" or "Chan 2".
    :type chan_id: str

    >>> rfgen = rpyc.connect('127.0.0.1', 18864)
    >>> rfgen.root.initialize()
    >>> chans = rfgen.root.channels

    >>> chan_id = 'Chan 1'
    >>> chans[chan_id].plo_ctl.freq = 1500.0
    >>> rfgen.root.configure_plo_freq(chan_id)

    """

def configure_plo(self, chan_id: str) -> None:
    """Update the PLO module hardware for the specified channel.

    Note that this should only be used when the generator is
    in calibration mode.

    .. seealso:: :meth:`.RFGenService.set_calibration_mode` and

    :param chan_id: A signal generator channel id.
        This will be one of "Chan 1" or "Chan 2".
    :type chan_id: str

    >>> rfgen = rpyc.connect('127.0.0.1', 18864)
    >>> rfgen.root.initialize()
    >>> chans = rfgen.root.channels

    >>> chan_id = 'Chan 1'
    >>> chans[chan_id].plo_ctl.buffer_gain = \
    ...     hmc833.OutputBufferGain.MAXGAIN_MINUS_6DB
    >>> rfgen.root.configure_plo(chan_id)

    """

def configure_atten(self, chan_id: str) -> None:
    """Update step attenuator module hardware for the specified channel.

    Note that this should only be used when the generator is
    in calibration mode.

    .. seealso:: :meth:`.RFGenService.set_calibration_mode` and

    :param chan_id: A signal generator channel id.
        This will be one of "Chan 1" or "Chan 2".
    :type chan_id: str

    >>> rfgen = rpyc.connect('127.0.0.1', 18864)
    >>> rfgen.root.initialize()
    >>> chans = rfgen.root.channels

    >>> chan_id = 'Chan 2'
    >>> chans[chan_id].atten_ctl.attenuation = 10.0
    >>> rfgen.root.configure_atten(chan_id)

    """

def detector(self, chan_id: str) -> Optional[PwrDetectorController]:
    """Return the controller associated with the power detector
    connected to the specified channel.

    :param chan_id: A signal generator channel id.
        This will be one of "Chan 1" or "Chan 2".
    :type chan_id: str

    :return: An instance of PwrDetectorController if there is a
        detector connected, None if no detector connected.
    """

def measure_power(self, chan_id: str) -> float:
    """Measure the signal power currently at the input of the power detector
    currently attached to the specified channel.

    :param chan_id: A signal generator channel id.
        This will be one of "Chan 1" or "Chan 2".
    :type chan_id: str

    :return: The measured signal power in dBm.

    :raises: KeyError
        If there is no detector currently connected to the specified
        channel.
    """

def set_channel_output(self, chan_id, freq, lvl, ignore_cal_state=False):
    """Set output frequency and power of the specified channel.

    :param chan_id: A signal generator channel id.
        This will be one of "Chan 1" or "Chan 2".
    :type chan_id: str
    :param freq: The output signal frequency (in MHz)
    :type freq: float
    :param lvl: The output signal power (in dBm)
    :type lvl: float
    :param ignore_cal_state: If this is set to False (the default value)
        an InvalidCalibrationDataError will be raised if the
        power detector connected to the associated channel monitoring
        output is not calibrated.  If set to True, the calibration
        state of the detector is ignored.
    :type ignore_cal_state: bool

    :return: True if the signal output was successfully set,
        False otherwise.

    :raises: KeyError
        If there is no detector currently connected to the specified
        channel.
    :raises: InvalidCalibrationDataError
        If the detector is not calibrated and the ignore_cal_status
        parameter is set to False.
    """

def configure(self, chan_id: str) -> None:
    """Update the channel hardware - PLO and attenuator.

    :param chan_id: A signal generator channel id.
        This will be one of "Chan 1" or "Chan 2".
    :type chan_id: str
    """

def set_calibration_mode(self, mode: bool) -> None:
    """Set the calibration mode.

    :param mode: True to set the app to calibration mode,
        False to set the app for normal operation.
    :type mode: bool

    Setting the calibration mode to True will disconnect the
    HMC833 controller ``freq_changed`` signal for each signal
    generator channel.  This then prevents the ``_update_channel_settings``
    method being invoked which in turn prevents any change to
    the channel attenuation.

    >>> rfgen = rpyc.connect('127.0.0.1', 18864)
    >>> rfgen.root.initialize()
    >>> rfgen.root.set_calibration_mode(True)

    """

@property
def enabled(self) -> bool:
    """Enable/disable the on screen PLO channel UI components.
    """

@property
def plo_ctl(self) -> HMC833Controller:
    """Return a reference to the channel's PLO controller.
    """

@property
def atten_ctl(self) -> PE43711Controller:
    """Return a reference to the channel's step attenuator
    controller.
    """

@property
def detector_ctl(self) -> Optional[PwrDetectorController]:
    """Return a reference to the channel's power detector controller.
    """

@property
def coupler(self) -> BridgeCoupler:
    """Return a reference to the channel's bridge coupler instance.
    """

@property
def level(self) -> float:
    """The current channel output power (in dBm).
    """

def configure(self):
    """Configure currently set frequency and level values in channel hardware.
    """

def power_range(self, freq):
    """Return the range of output signal powers at the specified
    frequency.

    :param freq: The output signal frequency
    :type freq: float

    :returns: A tuple containing the min and max output power
        available at the specified frequency.
    """
def set_calibration_mode(self, mode: bool) -> None:
    """Set the channel calibration mode.

    :param mode: True to set the channel to calibration mode,
        False to set the channel for normal operation.
    :type mode: bool

    Setting the calibration mode to True will disconnect the
    HMC833 controller ``freq_changed`` signal and thus prevent
    the ``_update_channel_settings`` method being invoked.
    This in turn prevents any change to the channel attenuation.
    """

from rfblocks import BridgeCoupler

coupler = BridgeCoupler()

chan_id = 'Chan 1'
freq = 1200.0
chan_output_pwr = (rfgen.measure_power(chan_id, freq=freq) - coupler.coupling(freq)
                   + coupler.insertion_loss(freq))

import os
import sys
from threading import Thread, Event, Lock
import rpyc


def level_output(lock, event, rfgen, chan_id):
    while True:
        with lock:
            freq = output_freq
            lvl = output_lvl
        if not rfgen.set_channel_output(chan_id, freq, lvl):
            print('Failed to set output level')
            break
        if event.is_set():
            print('level_output thread terminated')
            break


refdesigns_root = os.environ['RFBLOCKS_REFDESIGNS']
cal_file = f'{refdesigns_root}/PowerMeter/app/cal-files/24dB-atten-48-0007-Det3-100-6000MHz.json'
rfgen = rpyc.connect('127.0.0.1', 18864)
rfgen.root.initialize()
chan_id = 'Chan 1'
if not rfgen.root.initialize_detector(chan_id):
    print(f"Failed to initialize power detector for '{chan_id}'")
    sys.exit(-1)
det = rfgen.root.detector(chan_id)
if det is not None:
    det.load_caldata(cal_file)
    det.apply_correction = True
    det.avg = 4
else:
    print(f"Can't access power detector for '{chan_id}'")
    sys.exit(-2)

lock = Lock()
stop_event = Event()
output_freq = 1500.0
output_lvl = 0.0

level_thread = Thread(
    target=level_output,
    args=(lock, stop_event, rfgen.root, chan_id))
level_thread.start()

with lock:
    output_freq = 1200.0
    output_lvl = -5.0

# Stop the level thread
stop_event.set()

# Wait for the level thread to complete
level_thread.join()

import json
from argparse import ArgumentParser
import numpy as np


def format_cal_data_dict(cd_dict):
    cols = 4
    sd = [f'{f}: {float(p):.2f},' for f, p in cd_dict.items()]
    output_str = '\n'.join([''.join(['        '] + [f'{ls: <16}'
                            for ls in lst[:-1]] + [f'{lst[-1]}'])
                            for lst in [sd[i:i+cols]
                                        for i in range(0, len(sd), cols)]])
    return output_str


def format_chan_cal_data(chan_dict):
    output_str = '\n'.join([f"    '{k}': {{\n{format_cal_data_dict(cd_dict)}\n    }},"
                            for k, cd_dict in chan_dict.items()])
    return output_str


def read_cal_data(data_files):

    with open(data_files[0]) as fd:
        cal_data = json.load(fd)

    if len(data_files) > 1:
        # Read all the files and generate averages for the
        # measured signal powers.
        cal_data_list = [cal_data]
        for data_file in data_files[1:]:
            with open(data_file) as fd:
                cal_data_list.append(json.load(fd))
        chan_arrays = {}
        for data in cal_data_list:
            for chan_id, chan_dict in data.items():
                if chan_id in chan_arrays:
                    pwr_arrays = chan_arrays[chan_id]
                else:
                    pwr_arrays = {}
                    chan_arrays[chan_id] = pwr_arrays
                for gain_label, cd_dict in chan_dict.items():
                    parr = np.array([p for p in cd_dict.values()])
                    if gain_label in pwr_arrays:
                        pwr_arrays[gain_label] += parr
                    else:
                        pwr_arrays[gain_label] = parr
        for chan_id, chan_array in chan_arrays.items():
            for gain_label, pwr_array in chan_array.items():
                # Replace pwr values in cal_data with the averages
                # computed from pwr_array
                parr = chan_arrays[chan_id][gain_label] / len(data_files)
                cal_data[chan_id][gain_label] = {
                    fs: p for fs, p in zip(cal_data[chan_id][gain_label], parr)}

    return cal_data


def main(cal_data_files):

    cal_data = read_cal_data(cal_data_files)

    print('CHAN1_CAL_DATA = {')
    print(format_chan_cal_data(cal_data['Chan 1']))
    print('}')
    print('CHAN2_CAL_DATA = {')
    print(format_chan_cal_data(cal_data['Chan 2']))
    print('}')

if __name__ == '__main__':

    defaultInputFile = 'rfgen-output-power.json'

    parser = ArgumentParser(description=
                            '''Format RF signal generator calibration data''')
    parser.add_argument(
        "-I", "--input_files", default=[defaultInputFile], nargs='+',
        help=f"File containing calibration data as written by"
        " the rfgen-calibrate.py script."
        f" Default: {defaultInputFile}"
    )
    args = parser.parse_args()

    main(args.input_files)

from pathlib import Path
import skrf as rf
from skrf import Network, Frequency
import numpy as np
import matplotlib.pyplot as plt
from scipy.optimize import curve_fit
from dyadic.splot import init_style

def linear_func(x, a, b):
    return a*x + b

annotation_color = 'darkgrey'
freq_range = '10-6000mhz'
dataPath = Path(data_path)
ntwk = Network(str(dataPath / s12_file))[freq_range]
popt, perr = curve_fit(linear_func, ntwk.f/1e6, ntwk.s21.s_db[:,0,0])
print("coeffs: {},\nerrors: {}".format(popt, perr))
init_style()
fig = plt.figure(num=None, figsize=(6.0, 4.0), dpi=72)
ax = fig.add_subplot(111)
_ = ax.set_ylim(-4.0, -1.0)
_ = ax.set_xlim(0.0, 6000.0)
_ = ax.set_xlabel('Frequency (MHz)')
_ = ax.set_ylabel('Loss (dB)')
ax.grid(linestyle=':')
ax.grid(which='both', axis='x', linestyle=':')
ax.set_title('Insertion Loss Curve Fit')
ax.plot(ntwk.f/1e6, ntwk.s21.s_db[:,0,0])
ax.plot(ntwk.f/1e6, [linear_func(f, *popt) for f in ntwk.f/1e6],
        linestyle='dashed')
fig.tight_layout()
plt.savefig(outfile)

from pathlib import Path
import skrf as rf
from skrf import Network, Frequency
import numpy as np
import matplotlib.pyplot as plt
from scipy.optimize import curve_fit
from dyadic.splot import init_style

def poly_func(x, a, b, c, d, e, f, g):
    return (((((a*x + b)*x + c)*x + d)*x + e)*x + f)*x + g

annotation_color = 'darkgrey'
freq_range = '10-6000mhz'
dataPath = Path(data_path)
ntwk = Network(str(dataPath / s23_file))[freq_range]
popt, perr = curve_fit(poly_func, ntwk.f/1e6, ntwk.s21.s_db[:,0,0])
print("coeffs: {},\nerrors: {}".format(popt, perr))
init_style()
fig = plt.figure(num=None, figsize=(6.0, 4.0), dpi=72)
ax = fig.add_subplot(111)
_ = ax.set_ylim(-18.0, -15.0)
_ = ax.set_xlim(0.0, 6000.0)
_ = ax.set_xlabel('Frequency (MHz)')
_ = ax.set_ylabel('Coupling (dB)')
ax.grid(linestyle=':')
ax.grid(which='both', axis='x', linestyle=':')
ax.set_title('Coupling Curve Fit')
ax.plot(ntwk.f/1e6, ntwk.s21.s_db[:,0,0])
ax.plot(ntwk.f/1e6, [poly_func(f, *popt) for f in ntwk.f/1e6],
        linestyle='dashed')
fig.tight_layout()
plt.savefig(outfile)