Open RF Prototyping

# Generated from pwrmeter.org
#

<<imports>>

<<app-class>>

<<pwrmeter-service>>

<<main-func>>

if __name__ == '__main__':
    main()

def main():
    global server_thread

    <<process-cmdline-args>>

    app = QApplication(sys.argv)
    loop = QEventLoop(app)
    loop.set_default_executor(QThreadExecutor(1))
    asyncio.set_event_loop(loop)

    pwrmeter_app = PwrMeterApp(
        chan_cspins = [args.chan0_cspin, args.chan1_cspin],
        atten_lepins = [args.atten0_lepin, args.atten1_lepin],
        serial_device = args.device,
        baudrate = args.baudrate,
        headless = args.nogui)

    if not args.nogui:
        pwrmeter_app.show()

    server_thread = QThread()
    server = RPyCServer(PwrMeterService(pwrmeter_app),
                        args.ipaddr,
                        args.port)
    server.moveToThread(server_thread)
    server_thread.started.connect(server.run)
    server.finished.connect(server_thread.quit)
    server.finished.connect(server.deleteLater)
    server_thread.finished.connect(server_thread.deleteLater)
    server_thread.start()

    with loop:
        loop.set_debug(True)
        sys.exit(loop.run_forever())

class PwrMeterApp(QMainWindow):

    CALSTATUS_INCAL = 0
    CALSTATUS_ESTIMATED = 1
    CALSTATUS_UNCAL = 2

    CALSTATUS_STR = {
        CALSTATUS_INCAL: 'INCAL',
        CALSTATUS_ESTIMATED: 'ESTIMATED',
        CALSTATUS_UNCAL: 'UNCAL'}
    CALSTATUS_COLOURS = {
        CALSTATUS_INCAL: 'black',
        CALSTATUS_ESTIMATED: 'orange',
        CALSTATUS_UNCAL: 'red'
    }

    calstatus_changed = pyqtSignal(int, int)

    def __init__(self,
                 chan_cspins: List = None,
                 atten_lepins: List = None,
                 serial_device: Optional[str] = None,
                 baudrate: int = 0,
                 headless: bool = False) -> None:
        """
        """
        super().__init__()
        self._nogui: bool = headless
        self._detectors: List[PwrDetectorController] = [
            PwrDetectorController('0', LogDetector(chan_cspins[0])),
            PwrDetectorController('1', LogDetector(chan_cspins[1]))]
        self._attenuators: List[Optional[PE43711Controller]] = [None, None]
        if atten_lepins:
            for att_id in[0, 1]:
                if atten_lepins[att_id]:
                    att_ctl = PE43711Controller(
                        str(att_id), pe43711(atten_lepins[att_id]))
                    att_ctl.attenuation_changed.connect(
                        self.attenuation_changed)
                    self._attenuators[att_id] = att_ctl
        self._continuous: bool = False
        self._acquiring: bool = False
        self._shuttingdown: bool = False
        self._acquire_group_box: Optional[QGroupBox] = None
        self._ctl_device: str = serial_device
        self._baudrate: int = baudrate
        if baudrate == 0:
            self._baudrate = DEFAULT_BAUDRATE
        self._calstatus: List[int] = [
            PwrMeterApp.CALSTATUS_UNCAL, PwrMeterApp.CALSTATUS_UNCAL]
        for det_id, ctl in enumerate(self._detectors):
            ctl.caldata_changed.connect(partial(self.caldata_updated, det_id))
        self.calstatus_changed.connect(self.update_calstatus)

        if not headless:
            self._detector_displays: List[Optional[PowerDetector]] = [
                PowerDetector(ctl) for ctl in self.detectors]
            self._attenuator_displays: List[Optional[StepAttenuator]] = [None, None]
            for att_id in [0, 1]:
                att = self.attenuators[att_id]
                if att:
                    self._attenuator_displays[att_id] = StepAttenuator(
                        self, att)
            self.init_ui()

    def chan0_power_offset_fn(self, freq):
        """
        """
        return self._coupling + self._attn_insertion_loss

    def chan1_power_offset_fn(self, freq):
        """
        """
        return self._amplifier_gain + self._coupling

    @property
    def ctl_device(self) -> str:
        return self._ctl_device

    @property
    def serial_ports(self) -> List:
        ports = [DEFAULT_SOCKET_URL]
        ports.extend([port.device for port in comports() if port.serial_number])
        if self.ctl_device:
            if self.ctl_device in ports:
                ports.pop(ports.index(self.ctl_device))
            ports.insert(0, self.ctl_device)
        return ports

    @property
    def baudrate(self) -> int:
        return self._baudrate

    @property
    def detectors(self) -> List[PwrDetectorController]:
        """Returns a list containing the power detector controllers.
        """
        return self._detectors

    @property
    def attenuators(self) -> List[Optional[PE43711Controller]]:
        """Returns a list containing the step attenuator controllers.
        """
        return self._attenuators


    @property
    def calstatus(self) -> List[int]:
        """Returns a list containing the status of the detectors.
        """
        return self._calstatus

    @pyqtSlot(object)
    def attenuation_changed(self, att_ctl: PE43711Controller) -> None:
        """Device attenuation setting has changed.
        """
        det_id = int(att_ctl._controller_id)
        ctl: PwrDetectorController = self.detectors[det_id]

        # Check the calibration attenuation setting.
        # If there is no calibration attenuation setting
        # then just return without doing anything
        try:
            cal_atten = ctl.cal_data_conditions[
                PwrDetectorController.ATTEN_CAL_CONDITION]
        except KeyError:
            # There's no calibration attenuation value so
            # just set the insertion loss offset
            ctl.insertion_loss_offset = att_ctl.attenuation
            return

        # If the current attenuation setting for att_ctl
        # is different from the calibration attenuation
        # then calculate the insertion loss offset.
        loss_offset: float = att_ctl.attenuation - cal_atten
        if ctl.insertion_loss_offset == 0.0 and loss_offset != 0.0:
            # If the current value of the detector
            # controller insertion_loss_offset is zero and
            # the calculated loss offset is non-zero
            # emit a calstatus_changed signal.
            self.calstatus_changed.emit(det_id, PwrMeterApp.CALSTATUS_ESTIMATED)
        elif ctl.insertion_loss_offset != 0.0 and loss_offset == 0.0:
            # Likewise, if the detector controller
            # insertion_loss_offset is non-zero and the
            # calculated loss offset is zero
            # emit a calstatus_changed signal.
            self.calstatus_changed.emit(det_id, PwrMeterApp.CALSTATUS_INCAL)
        # Set the detector controller insertion loss offset
        ctl.insertion_loss_offset = loss_offset

    def disable_chan_controls(self, chan_id: int, f: bool) -> None:
        if chan_id in [0, 1]:
            self._detector_displays[chan_id].disable_controls(f)
            att_display = self._attenuator_displays[chan_id]
            if att_display:
                att_display.disable_controls(f)

    def disable_controls(self, f: bool) -> None:
        for det_id in [0, 1]:
            self.disable_chan_controls(det_id, f)
        if self._acquire_group_box:
            self._acquire_group_box.setDisabled(f)

    def initialize_detectors(self) -> None:
        with create_serial(self.ctl_device,
                           self.baudrate) as ser:
            for ctl in self.detectors:
                ctl.initialize(ser)

    @pyqtSlot()
    def caldata_updated(self, det_id):
        self.update_calstatus(det_id, PwrMeterApp.CALSTATUS_INCAL)

    @pyqtSlot(int, int)
    def update_calstatus(self, det_id: int, status: int):
        """
        """
        self.calstatus[det_id] = status
        if not self._nogui:
            grp: PowerDetector = self._detector_displays[det_id]
            grp._cal_status_lbl.setText(PwrMeterApp.CALSTATUS_STR[status])
            grp._cal_status_lbl.setStyleSheet(
                'QLabel { color: %s }' % PwrMeterApp.CALSTATUS_COLOURS[status])

    @property
    def continuous(self) -> bool:
        return self._continuous

    @property
    def acquiring(self) -> bool:
        return self._acquiring

    def set_continuous(self, c: bool) -> None:
        self._continuous = c

    def set_acquiring(self, a: bool) -> None:
        self._acquiring = a

    @property
    def shuttingdown(self) -> bool:
        return self._shuttingdown

    def shutdown(self, b: bool) -> None:
        self._shuttingdown = b

    <<hwconfig-funcs>>

    <<acquisition-funcs>>

    <<calibration>>

    def init_ui(self) -> None:
        vbox = QVBoxLayout()
        vbox.addWidget(self.create_hwconfig_group())

        hbox = QHBoxLayout()
        for det_id in [0, 1]:
            inner_vbox = QVBoxLayout()
            grp = self._detector_displays[det_id].create_detector_group(
                partial(self.calibrate, det_id))
            inner_vbox.addWidget(grp)
            att_display = self._attenuator_displays[det_id]
            if att_display:
                grp = att_display.create_attenuator_group()
                inner_vbox.addWidget(grp)
            else:
                inner_vbox.addStretch()
            hbox.addLayout(inner_vbox)
        self.disable_controls(True)
        vbox.addLayout(hbox)

        hbox2 = QHBoxLayout()
        grp = self.create_acquisition_control_group()
        hbox2.addWidget(grp)
        grp.setDisabled(True)
        vbox.addLayout(hbox2)

        self.container = QWidget()
        self.container.setLayout(vbox)

        self.setCentralWidget(self.container)

        self.setWindowTitle('Dual Detector Power Meter')

class PwrMeterService(rpyc.Service):

    def __init__(self, app):
        super().__init__()
        self._app = app

    def initialize(self) -> None:
        """Initialize the service app and associated hardware modules.
        """
        with create_serial(self._app.ctl_device,
                           self._app.baudrate) as ser:
            for ctl in self.detectors:
                ctl.initialize(ser)
        if not self._app._nogui:
            self._app.hw_config_btn.setEnabled(True)
            self._app.disable_controls(False)


    @property
    def detectors(self):
        """Return a list of the power detector objects.

        :return: A list of PwrDetectorController instances one for
        each of the power meter channels.
        """
        return self._app.detectors

    @property
    def attenuators(self):
        """Return a list of channel attenuator objects.
        Note that one of more of the objects may be ``None`` depending
        on whether the corresponding channel is actually equipped with
        a step attenuator.

        :return: A list of PE43711Controller instances one for each of
        the power meter channels.
        """
        return self._app.attenuators

    def set_attenuation(self, ctl_id, atten):
        """Set the attenuation for the specified power meter channel.

        :param ctl_id: Power meter channel id.
        :type ctl_id: int
        :param atten: Required attenuation level.
        :type atten: float

        Note that this will only have any effect if the channel is
        actually equipped with a step attenuator.
        """
        att = self.attenuators[ctl_id]
        if att:
            if not self._app._nogui:
                grp = self._app._attenuator_displays[ctl_id]
                if grp:
                    grp._atten_box.setValue(atten)
            else:
                att.attenuation = atten
            with create_serial(self._app.ctl_device,
                               self._app.baudrate) as ser:
                att.configure(ser)

    def detector_enable(self, ctl_id, en):
        """Enable/disable the specified power meter channel.

        :param ctl_id: Power meter channel id.
        :type ctl_id: int
        :param en: True to enable the channel, False disables the channel.
        :type en: bool
        """
        if not self._app._nogui:
            state = Qt.Checked if en else Qt.Unchecked
            self._app._detector_displays[ctl_id].set_enabled(state)
            att_display = self._app._attenuator_displays[ctl_id]
            if att_display is not None:
                att_display.disable_controls(en)
        else:
            ctl = self.detectors[ctl_id].enabled = en

    def calibrate(self, ctl_id, src_dev, device_id, src_chan,
                  srcpwr, startf, stopf, stepf):
        """Carry out a power meter channel calibration.

        :param ctl_id: Power meter channel id.
        :type ctl_id: int
        :param src_dev: Calibration signal source device.
        :type src_dev: str
        :param device_id: Calibration signal source device id.
        :type device_id: str
        :param src_chan: Calibration signal source channel.
        :type src_chan: int
        :param srcpwr: Calibration signal source power level (in dBm).
        :type srcpwr: float
        :param startf: Calibration start frequency (in MHz).
        :type startf: float
        :param stopf: Calibration stop frequency (in MHz).
        :type stopf: float
        :param stepf: Calibration frequency step size (in MHz)
        :type stepf: float

        .. code:: python
            import rpyc
            from tam import DSG815_ID

            pwrmeter = rpyc.connect('127.0.0.1', 18863)
            pwrmeter.root.initialize()
            chan_controllers = pwrmeter.root.detectors
            chan_ctl0 = chan_controllers[0]
            chan_ctl1 = chan_controllers[1]

            # Calibration is a long running process and the calibrate call
            # is synchronous.  The rpyc request timeout should be either
            # disabled or set to a long period so that a timeout isn't triggered
            old_timeout = pwrmeter._config['sync_request_timeout']
            pwrmeter._config['sync_request_timeout'] = None
            status = pwrmeter.root.calibrate(0, 'DSG815', DSG815_ID, 0, 0.0, 100.0, 1500.0, 100.0)
            pwrmeter.root.calstatus[0]   # should return 0

            # Restore the rpyc request timeout value
            pwrmeter._config['sync_request_timeout'] = old_timeout

        .. seealso::

            :py:meth:`calstatus`
        """
        return self._app.calibrate(
            ctl_id, src_dev, device_id, src_chan,
            srcpwr, startf, stopf, stepf)

    def save_caldata(self, ctl_id, cal_data_file):
        """Save channel calibration data to the specified file.

        :param ctl_id: Power meter channel id.
        :type ctl_id: int
        :param cal_data_file: Path to file where cal. data will be saved.
        :type cal_data_file: str
        """
        ctl = self.detectors[ctl_id]
        ctl.save_caldata(cal_data_file)

    def load_caldata(self, ctl_id, cal_data_file):
        """Load channel calibration data from the specified file.

        :param ctl_id: Power meter channel id.
        :type ctl_id: int
        :param cal_data_file: Path to file from which cal. data will be loaded.
        :type cal_data_file: str
        """
        ctl = self.detectors[ctl_id]
        ctl.load_caldata(cal_data_file)

    @property
    def calstatus(self):
        """Return the status of the most recent calibration operation.

        :return: True if the calibration completed successfully,
        False otherwise.

        .. seealso::

            :py:meth:`calibrate`
        """
        return self._app.calstatus

    def measure(self, ctl_id: int) -> None:
        """Take a power measurement on the specified power meter channel.

        :param ctl_id: Power meter channel id.
        :type ctl_id: int

        .. code:: python
            import rpyc

            pwrmeter = rpyc.connect('127.0.0.1', 18863)
            pwrmeter.root.initialize()
            chan_controllers = pwrmeter.root.detectors
            chan_ctl0 = chan_controllers[0]
            chan_ctl1 = chan_controllers[1]

            chan_ctl0.freq = 500.0
            pwrmeter.root.measure(0)
            pwr = chan_ctl0.pwr
            chan_ctl0.apply_correction = True
            pwrmeter.root.measure(0)
            corrected_pwr = chan_ctl0.pwr

            chan_ctl1.freq = 150.0
            pwrmeter.root.measure(1)
            pwr = chan_ctl1.pwr

            pwrmeter.close()
        """
        with create_serial(self._app.ctl_device,
                           self._app.baudrate) as ser:
            self.detectors[ctl_id].measure(ser)


class RPyCServer(QObject):

    finished = pyqtSignal()

    def __init__(self, serviceInst, host, port):
        super().__init__()
        self._serviceInst = serviceInst
        self._host = host
        self._port = port

    def run(self):
        print("PwrMeter rpyc service on {}:{}".format(self._host, self._port))
        self._server = ThreadedServer(
            self._serviceInst,
            hostname = self._host,
            port = self._port,
            protocol_config = {
                'allow_all_attrs': True,
                'allow_setattr': True,
                'allow_pickle': True})
        self._server.start()
        self.finished.emit()

from typing import (
    Optional, List, Dict, Tuple
)

from functools import (
    partial
)

import sys
import traceback
from time import sleep
import asyncio
import numpy as np
from argparse import ArgumentParser
import serial
from serial.tools.list_ports import comports

from rfblocks import (
    LogDetector, PwrDetectorController, pe43711, PE43711Controller,
    DetectorReadError, create_serial, write_cmd, query_cmd,
    DEFAULT_BAUDRATE, DEFAULT_SOCKET_URL
)

from qtrfblocks import (
    PowerDetector, StepAttenuator
)

from qasync import (
    QEventLoop, QThreadExecutor, asyncSlot, asyncClose
)

import pyvisa

from tam import (
    InstrumentMgr
)

from PyQt5.QtWidgets import (
    QApplication, QWidget, QMainWindow, QLabel, QPushButton,
    QVBoxLayout, QHBoxLayout, QFormLayout, QComboBox, QGroupBox,
    QErrorMessage, QDoubleSpinBox, QLineEdit, QButtonGroup, QRadioButton,
    QCheckBox
)

from PyQt5.QtCore import (
    Qt, QObject, QThread, pyqtSignal, pyqtSlot
)

import rpyc
from rpyc.utils.server import ThreadedServer