feat: Ch05 Digital Modulation

Signed-off-by: Philipp Le <philipp@philipple.de>
Change-Id: Idec2fd38a0da8ddd2dc9292a0dbc6628ca427cbe
diff --git a/dcs/frames/ch05_qam.py b/dcs/frames/ch05_qam.py
new file mode 100644
index 0000000..448f37f
--- /dev/null
+++ b/dcs/frames/ch05_qam.py
@@ -0,0 +1,348 @@
+#  SPDX-License-Identifier: MPL-2.0
+#    Copyright (c) 2022 Philipp Le <philipp@philipple.de>.
+#  This Source Code Form is subject to the terms of the Mozilla Public
+#  License, v. 2.0. If a copy of the MPL was not distributed with this
+#  file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+from __future__ import annotations
+
+from tkinter import ttk, LEFT, BOTH, BOTTOM
+from pydantic import confloat, conint
+from dcs.config import default_store, ConfigObject, ui_create, ConfigControlFrame
+import numpy as np
+import scipy.signal
+from dcs.frames.base import BaseFrame, Window
+from dcs.frames.groups import Ch05Group
+from dcs.utils.filter import create_lp_filter_class
+from dcs.utils.iq_mixer import create_iq_oscillator_class, IqUpMixer, IqDownMixer, IqRfBandGenerator, IqRxBasebandGenerator
+from dcs.utils.qam_modulation import ModulationMethod, QamBasebandModulator, QamBasebandGenerator
+from dcs.utils.symbols import MessageBinarySerializer
+from typing import  Dict
+
+import matplotlib
+matplotlib.use('TkAgg')
+from matplotlib.pyplot import Circle, Text
+from matplotlib.figure import Figure
+from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg, NavigationToolbar2Tk
+
+
+SAMPLE_RATE = 10000
+FFT_SAMPLES_PER_SYM = 64
+
+
+IqOscillator = create_iq_oscillator_class(sample_rate=SAMPLE_RATE, imbalance_wdg=True)
+
+TxBasebandFilter = create_lp_filter_class(SAMPLE_RATE, 'TX Baseband')
+RxBasebandFilter = create_lp_filter_class(SAMPLE_RATE, 'RX Baseband')
+
+
+class ConfigCh05QamFrame(ConfigControlFrame):
+    wdg_msg_update_btn: ttk.Button
+
+
+@ui_create
+class ConfigCh05Qam(ConfigObject):
+    _KEY = 'ch05_qam'
+
+    message: str = 'DCS'
+    symbol_rate: confloat(ge=-SAMPLE_RATE/4, lt=SAMPLE_RATE/4, multiple_of=(4.0/SAMPLE_RATE)) = 10.0
+    modulation: ModulationMethod = ModulationMethod.QPSK
+    tx_lp: TxBasebandFilter = TxBasebandFilter()
+    tx_osc: IqOscillator = IqOscillator(freq=100.0)
+    noise_dBc: confloat(ge=-160.0, le=40.0, multiple_of=0.1) = -60.0
+    random_seed: conint(ge=0, lt=10000) = 1000
+    rx_osc: IqOscillator = IqOscillator(freq=100.0)
+    rx_lp: RxBasebandFilter = RxBasebandFilter()
+
+    def _mod_frame(self, parent: ConfigCh05QamFrame) -> ConfigControlFrame:
+        frm = ConfigControlFrame(parent, borderwidth=1, relief='raised')
+
+        ttk.Label(frm, text='Message:').grid(row=0, column=0)
+        # NOTE: Recalculating the message is slow. So only update on click on the Update button.
+        self.ui_create_message(frm).grid(row=0, column=1)
+        parent.wdg_msg_update_btn = ttk.Button(frm, text='Update')
+        parent.wdg_msg_update_btn.grid(row=0, column=2)
+
+        ttk.Label(frm, text='Symbol Rate:').grid(row=1, column=0)
+        w = self.ui_create_symbol_rate(frm)
+        frm.add_widget(w)
+        w.grid(row=1, column=1)
+        ttk.Label(frm, text='Hz').grid(row=1, column=2)
+
+        ttk.Label(frm, text='Modulation:').grid(row=2, column=0)
+        w = self.ui_create_modulation_dropdown(frm)
+        frm.add_widget(w)
+        w.grid(row=2, column=1)
+
+        ttk.Label(frm, text='TX Baseband Low Pass Cut-Off:').grid(row=3, column=0)
+        w = self.tx_lp.ui_create_cutoff(frm)
+        frm.add_widget(w)
+        w.grid(row=3, column=1)
+        ttk.Label(frm, text='Hz').grid(row=3, column=2)
+
+        return frm
+
+    def _noise_frame(self, parent: ttk.Widget) -> ConfigControlFrame:
+        frm = ConfigControlFrame(parent, borderwidth=1, relief='raised')
+
+        ttk.Label(frm, text='Signal-Noise Ratio:').grid(row=0, column=0)
+        w = self.ui_create_noise_dBc(frm)
+        frm.add_widget(w)
+        w.grid(row=0, column=1)
+        ttk.Label(frm, text='dBc').grid(row=0, column=2)
+
+        ttk.Label(frm, text='Random Seed:').grid(row=1, column=0)
+        w = self.ui_create_random_seed(frm)
+        frm.add_widget(w)
+        w.grid(row=1, column=1)
+
+        return frm
+
+    def _demod_frame(self, parent: ttk.Widget) -> ConfigControlFrame:
+        frm = ConfigControlFrame(parent, borderwidth=1, relief='raised')
+
+        frm1 = self.rx_lp.make_config_widget(frm)
+        frm1.pack()
+        for w in frm1.ctrl_widgets:
+            frm.add_widget(w)
+
+        return frm
+
+    def make_config_widget(self, parent: ttk.Widget) -> ConfigCh05QamFrame:
+        frm = ConfigCh05QamFrame(parent, borderwidth=1, relief='raised')
+
+        sym_frm = self._mod_frame(frm)
+        sym_frm.pack()
+        for w in sym_frm.ctrl_widgets:
+            frm.add_widget(w)
+
+        tx_frm = ttk.Frame(frm, borderwidth=1, relief='raised')
+        tx_frm.pack()
+        ttk.Label(tx_frm, text='TX Oscillator:').pack()
+        tx_ctrl_frm = self.tx_osc.make_config_widget(tx_frm)
+        tx_ctrl_frm.pack()
+        for w in tx_ctrl_frm.ctrl_widgets:
+            frm.add_widget(w)
+
+        noise_frm = self._noise_frame(frm)
+        noise_frm.pack()
+        for w in noise_frm.ctrl_widgets:
+            frm.add_widget(w)
+
+        rx_frm = ttk.Frame(frm, borderwidth=1, relief='raised')
+        rx_frm.pack()
+        ttk.Label(rx_frm, text='RX Oscillator:').pack()
+        rx_ctrl_frm = self.tx_osc.make_config_widget(rx_frm)
+        rx_ctrl_frm.pack()
+        for w in rx_ctrl_frm.ctrl_widgets:
+            frm.add_widget(w)
+
+        demod_frm = self._demod_frame(frm)
+        demod_frm.pack()
+        for w in demod_frm.ctrl_widgets:
+            frm.add_widget(w)
+
+        return frm
+
+
+class Ch05QamFrame(BaseFrame):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        self._config: ConfigCh05Qam = default_store().get_config(ConfigCh05Qam)
+
+        ctrl_frm = self._create_control()
+        ctrl_frm.pack(side=LEFT)
+
+        signal_frm = self._create_signal_tabs()
+        signal_frm.pack(expand=True, fill=BOTH)
+
+    def _create_control(self) -> ConfigControlFrame:
+        frm = self._config.make_config_widget(self)
+        frm.widgets_on_change(self._on_change)
+        frm.wdg_msg_update_btn.configure(command=lambda: self._on_change(0, 0, 0))
+        return frm
+
+    def _on_change(self, _, __, ___):
+        default_store().save()
+        self.refresh()
+
+    def _create_signal_tabs(self) -> ttk.Widget:
+        tabs = ttk.Notebook(self)
+
+        tx_frm = ttk.Frame(tabs)
+        tabs.add(tx_frm, text='TX Baseband')
+        self._tx_fig = Figure(figsize=(12, 6), dpi=100)
+        self._tx_canvas = FigureCanvasTkAgg(self._tx_fig, tx_frm)
+        self._tx_canvas.get_tk_widget().pack(expand=True, fill=BOTH)
+        tx_tb = NavigationToolbar2Tk(self._tx_canvas, tx_frm, pack_toolbar=False)
+        tx_tb.pack(side=BOTTOM)
+
+        self._data_bytes = ttk.Label(tx_frm, text='')
+        self._data_bytes.pack(side=BOTTOM)
+        self._data_syms = ttk.Label(tx_frm, text='')
+        self._data_syms.pack(side=BOTTOM)
+
+        hf_frm = ttk.Frame(tabs)
+        tabs.add(hf_frm, text='HF Band')
+        self._hf_fig = Figure(figsize=(12, 6), dpi=100)
+        self._hf_canvas = FigureCanvasTkAgg(self._hf_fig, hf_frm)
+        self._hf_canvas.get_tk_widget().pack(expand=True, fill=BOTH)
+        hf_tb = NavigationToolbar2Tk(self._hf_canvas, hf_frm, pack_toolbar=False)
+        hf_tb.pack(side=BOTTOM)
+
+        rx_frm = ttk.Frame(tabs)
+        tabs.add(rx_frm, text='RX Baseband')
+        self._rx_fig = Figure(figsize=(12, 6), dpi=100)
+        self._rx_canvas = FigureCanvasTkAgg(self._rx_fig, rx_frm)
+        self._rx_canvas.get_tk_widget().pack(expand=True, fill=BOTH)
+        rx_tb = NavigationToolbar2Tk(self._rx_canvas, rx_frm, pack_toolbar=False)
+        rx_tb.pack(side=BOTTOM)
+
+        self.refresh()
+
+        return tabs
+
+    def refresh(self):
+        # Make constellation
+        constellation = self._config.modulation.make_constellation_map_str()
+
+        # Make symbols vector
+        bytes_symbols = MessageBinarySerializer(self._config.message).to_symbols()
+        symbols = bytes_symbols.reencode(self._config.modulation.bits_per_symbol())
+
+        # Update symbol output widgets
+        data_bytes = ', '.join(bytes_symbols.to_bit_str())
+        self._data_bytes.configure(text=f'Data = [{data_bytes}]')
+        data_syms = ', '.join(symbols.to_bit_str())
+        self._data_syms.configure(text=f'Symbols = [{data_syms}]')
+
+        # Create signal generators
+        baseband_gen = QamBasebandGenerator(
+            symbols,
+            QamBasebandModulator(
+                method=self._config.modulation,
+                baseband_filter=self._config.tx_lp,
+                symbol_rate=self._config.symbol_rate
+            )
+        )
+        iq_mod = baseband_gen.make_rf_band_generator(
+            IqUpMixer(
+                tx_osc=self._config.tx_osc,
+                noise_dBc=self._config.noise_dBc,
+                random_seed=self._config.random_seed
+            )
+        )
+        iq_demod = iq_mod.make_iq_down_mixer(
+            IqDownMixer(
+                rx_osc=self._config.rx_osc,
+                baseband_filter=self._config.rx_lp
+            )
+        )
+
+        # Draw
+        self.draw_tx(constellation, baseband_gen)
+        self.draw_hf(iq_mod)
+        self.draw_rx(constellation, iq_demod)
+
+    def draw_tx(self, constellation: Dict[str, complex], baseband_gen: QamBasebandGenerator):
+        self._tx_fig.clear()
+
+        ax_const = self._tx_fig.add_subplot(2, 1, 1)
+        ax_const.set_xlim(-1.2, 1.2)
+        ax_const.set_ylim(-1.2, 1.2)
+        ax_const.set_xlabel('real')
+        ax_const.set_ylabel('imaginary')
+        ax_const.set_title('Ideal Constellation Diagram')
+        ax_const.set_aspect(1)
+        ax_base = self._tx_fig.add_subplot(2, 1, 2)
+        ax_base.set_xlabel('time')
+        ax_base.set_ylabel('value')
+        ax_base.set_title('TX Baseband Signal')
+
+        for sym_str, sym_val in constellation.items():
+            circ = Circle((sym_val.real, sym_val.imag), 0.03, alpha=0.4, color='blue', label=sym_str)
+            ax_const.add_artist(circ)
+            txt = Text(sym_val.real, sym_val.imag, sym_str, color='orange', fontsize=8)
+            ax_const.add_artist(txt)
+
+        sig_base_tx = baseband_gen.generate_tx_baseband_signal(SAMPLE_RATE)
+        ax_base.set_xlim(np.min(sig_base_tx.t), np.max(sig_base_tx.t))
+        ax_base.plot(sig_base_tx.t, np.real(sig_base_tx.signal), label='Baseband I', linestyle='solid', color='blue', linewidth=1)
+        ax_base.plot(sig_base_tx.t, np.imag(sig_base_tx.signal), label='Baseband Q', linestyle='solid', color='red', linewidth=1)
+        ax_base.legend()
+
+        self._tx_fig.tight_layout()
+        self._tx_canvas.draw()
+
+    def draw_hf(self, rf_gen: IqRfBandGenerator):
+        self._hf_fig.clear()
+
+        ax_td = self._hf_fig.add_subplot(2, 1, 1)
+        ax_td.set_xlabel('time')
+        ax_td.set_ylabel('value')
+        ax_td.set_title('HF Signal Time-Domain')
+        ax_fd = self._hf_fig.add_subplot(2, 1, 2)
+        ax_fd.set_ylim(-140, 40)
+        ax_fd.set_xlabel('frequency')
+        ax_fd.set_ylabel('value (logarithmic)')
+        ax_fd.set_title('HF Signal Frequency-Domain')
+
+        sig_hf_normal = rf_gen.generate_rf_signal(SAMPLE_RATE)
+        ax_td.set_xlim(np.min(sig_hf_normal.t), np.max(sig_hf_normal.t))
+        ax_td.plot(sig_hf_normal.t, np.real(sig_hf_normal.signal), label='HF', linestyle='solid', color='brown', linewidth=1)
+        ax_td.legend()
+
+        fd_smp_rate = FFT_SAMPLES_PER_SYM * self._config.symbol_rate
+        sig_hf_ovs = rf_gen.generate_rf_signal(fd_smp_rate)
+        f_fd, X_hf_fd = sig_hf_ovs.fft(window_fn=scipy.signal.windows.blackman)
+
+        ax_fd.set_xlim(0, np.max(f_fd))
+        ax_fd.plot(f_fd, X_hf_fd, label='HF', linestyle='solid', color='brown', linewidth=1)
+        ax_fd.legend()
+
+        self._hf_fig.tight_layout()
+        self._hf_canvas.draw()
+
+    def draw_rx(self, constellation: Dict[str, complex], rx_gen: IqRxBasebandGenerator):
+        self._rx_fig.clear()
+        ax_base = self._rx_fig.add_subplot(2, 1, 1)
+        ax_base.set_xlabel('time')
+        ax_base.set_ylabel('value')
+        ax_base.set_title('RX Baseband Signal')
+        ax_const = self._rx_fig.add_subplot(2, 1, 2)
+        ax_const.set_xlim(-1.2, 1.2)
+        ax_const.set_ylim(-1.2, 1.2)
+        ax_const.set_xlabel('real')
+        ax_const.set_ylabel('imaginary')
+        ax_const.set_title('RX Constellation Diagram')
+        ax_const.set_aspect(1)
+
+        sig_base_rx = rx_gen.generate_rx_baseband_signal(SAMPLE_RATE)
+        ax_base.set_xlim(np.min(sig_base_rx.t), np.max(sig_base_rx.t))
+        ax_base.plot(sig_base_rx.t, np.real(sig_base_rx.signal), label='Baseband I', linestyle='solid', color='blue', linewidth=1)
+        ax_base.plot(sig_base_rx.t, np.imag(sig_base_rx.signal), label='Baseband Q', linestyle='solid', color='red', linewidth=1)
+        ax_base.legend()
+
+        ax_const.plot(np.real(sig_base_rx.signal), np.imag(sig_base_rx.signal), label='Baseband', linestyle='solid', color='purple', linewidth=1)
+        ax_const.legend()
+
+        for sym_str, sym_val in constellation.items():
+            circ = Circle((sym_val.real, sym_val.imag), 0.05, alpha=0.4, color='blue', label=sym_str)
+            ax_const.add_artist(circ)
+            txt = Text(sym_val.real, sym_val.imag, sym_str, color='orange', fontsize=8)
+            ax_const.add_artist(txt)
+
+        self._rx_fig.tight_layout()
+        self._rx_canvas.draw()
+
+
+class Ch05QamWindow(Window):
+    GROUP = Ch05Group
+    TITLE = 'Quadrature Amplitude Modulation'
+    FRAME = Ch05QamFrame
+
+
+if __name__ == '__main__':
+    Ch05QamWindow.main()
+