Open RF Prototyping

# Generated from clkgen.org
#

<<imports>>

<<msg-dialog>>

<<clockgen-app-class>>

<<clkgen-service>>

<<main-func>>

if __name__ == '__main__':
    main()

def main():
    global server_thread

    <<process-cmdline-args>>

    # This ensures that Cntl-C will work as expected:
    signal.signal(signal.SIGINT, signal.SIG_DFL)

    if args.nogui is True:
        app = QCoreApplication(sys.argv)
    else:
        app = QApplication(sys.argv)
    loop = QEventLoop(app)
    loop.set_default_executor(QThreadExecutor(1))
    asyncio.set_event_loop(loop)

    clkgen_app = ClkApp(serial_device=args.device,
                        baudrate=args.baudrate,
                        headless=args.nogui)
    if args.dumphw:
        for mod_id, controller in clkgen_app.controllers.items():
            print(f'{mod_id}: {vars(controller._ad9552)}')
        sys.exit(0)

    if not args.nogui:
        clkgen_app.show()

    server_thread = QThread()
    server = RPyCServer(ClkgenService(clkgen_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()

    sys.exit(app.exec_())

class ClkApp(QObject):

    DEFAULT_CLKOUT_FREQ = 250.0

    <<clkapp-config>>

    def __init__(self,
                 config: Dict = DEFAULT_APP_CONFIG,
                 serial_device: Optional[str] = None,
                 baudrate: int = 0,
                 headless: bool = False) -> None:
        super().__init__()
        self._nogui: bool = headless
        if self._nogui is False:
            self._widget = QWidget()
        d: Dict = config['modules']
        devices: List[ad9552] = [
            ad9552(**hw_conf) for hw_conf in ClkApp.MODULES_HWCONF]
        cz = zip(d.keys(), devices, d.values())
        self.controllers: Dict = {
            mod_id: AD9552Controller(mod_id, dev, mod_config)
            for mod_id, dev, mod_config in cz }
        self._ctl_device: Optional[str] = serial_device
        self._baudrate: int = baudrate
        if baudrate == 0:
            self._baudrate = DEFAULT_BAUDRATE

        if not headless:
            mz = zip(d.keys(), self.controllers.values(),
                     [[ch['ui'] for ch in mod['channels'].values()] \
                      for mod in d.values()])
            self._modules = {mod_id: ClkModule(self, mod_id, ctl, *has_ui)
                             for mod_id, ctl, has_ui in mz}
            self.build_ui()

    <<build-app-ui>>

    <<initialize>>

    <<app-configuration>>

    @property
    def ctl_device(self) -> Optional[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

    @asyncSlot(QAbstractButton)
    async def set_refsrc(self, btn) -> None:
        """Set the reference source for the generator.

        The reference source is controlled via the `Ref` pin
        on clock module 1.
        """
        clk_ctl = self.controllers[ClkApp.CLKMODULE1_NAME]
        if btn.src == clk_ctl.refsrc:
            return
        clk_ctl.refsrc = btn.src
        try:
            cmd = clk_ctl._ad9552.config_refsrc()
            if len(cmd):
                loop = asyncio.get_event_loop()
                with create_serial(self.ctl_device,
                                   self.baudrate) as ser:
                    await loop.run_in_executor(
                        None, write_cmd, ser, cmd)
        except serial.serialutil.SerialException as se:
            error_dialog = QErrorMessage(self)
            error_dialog.showMessage(str(se))

    def tty_changed(self, combo: QComboBox, idx: int) -> None:
        self._ctl_device = combo.itemText(idx)

    def disable_ui(self) -> None:
        for module in self._modules.values():
            module.enabled = False
        for btn in self._refsrc_btns:
            btn.setEnabled(False)

    def enable_ui(self) -> None:
        for module in self._modules.values():
            module.enabled = True
        for btn in self._refsrc_btns:
            btn.setEnabled(True)

class ClkgenService(rpyc.Service):

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

    def initialize(self) -> None:
        """Initialize the clock generator hardware and software.

        >>> import rpyc
        >>> clkgen = rpyc.connect("127.0.0.1", 18861)
        >>> clkgen.root.initialize()

        """
        if not self._app._nogui:
            self._app.initialize_ui()
        self._app.initialize_modules()
        if not self._app._nogui:
            self._app.enable_ui()

    @property
    def controllers(self) -> Dict[str, AD9552Controller]:
        """A dictionary containing the clock generator clock module controllers.

        The controllers are instances of AD9552Controller keyed using
        the clock module identifiers 'clkmod1' and 'clkmod2' respectively.

        >>> import rpyc
        >>> clkgen = rpyc.connect("127.0.0.1", 18861)
        >>> clkgen.root.initialize()
        >>> clkmod1 = clkgen.root.controllers['clkmod1']
        >>> clkmod2 = clkgen.root.controllers['clkmod2']

        """
        return self._app.controllers

    def configure(self, ctl_id: str) -> bool:
        """Update the clock module hardware using the currently set configuration.

        :param ctl_id: A module id.  This will be either 'clkmod1' or 'clkmod2'.
        :type ctl_id: str

        Returns a boolean value indicating the clock module PLL lock status.
        True for locked.

        >>> import rpyc
        >>> clkgen = rpyc.connect("127.0.0.1", 18861)
        >>> clkgen.root.initialize()
        >>> clkmod1 = clkgen.root.controllers['clkmod1']
        >>> clkmod1.freq = 150.0
        >>> chan1 = clkmod1.channels['1']
        >>> chan1.state = ad9552.OutputState.ACTIVE
        >>> pll_lock = clkgen.root.configure('clkmod1')

        """
        ctl = self._app.controllers[ctl_id]
        with create_serial(self._app.ctl_device,
                           self._app.baudrate) as ser:
            pll_lock = ctl.configure(ser)
        return pll_lock

    def configure_freq(self, ctl_id: str) -> bool:
        """Update the clock module hardware with the currently set
        output frequency.

        :param ctl_id: A module id.  This will be either 'clkmod1' or 'clkmod2'.
        :type ctl_id: str

        Returns a boolean value indicating the clock module PLL lock status.
        True for locked.

        >>> import rpyc
        >>> clkgen = rpyc.connect("127.0.0.1", 18861)
        >>> clkgen.root.initialize()
        >>> clkmod1 = clkgen.root.controllers['clkmod1']
        >>> clkmod1.freq = 150.0
        >>> clkgen.root.configure_freq('clkmod1')

        """
        ctl = self._app.controllers[ctl_id]
        with create_serial(self._app.ctl_device,
                           self._app.baudrate) as ser:
            pll_lock = ctl.configure_freq(ser)
        return pll_lock


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("ClockGen 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_getattr': True})
        self._server.start()
        self.finished.emit()

def popMsgDialog(icon, title, msg, info=''):
    """
    """
    msgBox = QMessageBox()
    msgBox.setIcon(QMessageBox.NoIcon)
    msgBox.setWindowTitle(title)
    msgBox.setText(msg)
    if len(info):
        msgBox.setInformativeText(info)
    msgBox.setStandardButtons(QMessageBox.Ok)
    status = msgBox.exec_()

from typing import (
    Optional, List, Dict
)
import sys
import signal
import asyncio
from argparse import ArgumentParser
import serial
import json
from serial.tools.list_ports import comports

from rfblocks import (
    ad9552, AD9552Controller,
    DividerRangeException, list_available_serial_ports,
    create_serial, write_cmd, query_cmd,
    DEFAULT_BAUDRATE, DEFAULT_SOCKET_URL
)

from qtrfblocks import (
    ClkModule, ClkChannel
)

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

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

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

import rpyc
from rpyc.utils.server import ThreadedServer