| # 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 |
| |
| import copy |
| 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 Ch07Group |
| from dcs.utils.chain import TxBasebandGenerator, RxCorrelationGenerator, RfBandGenerator |
| from dcs.utils.filter import create_lp_filter_class, NoFilter |
| from dcs.utils.iq_mixer import create_iq_oscillator_class, IqUpMixer, IqDownMixer, IqRfBandGenerator, \ |
| IqRxBasebandGenerator, IqTxBasebandGenerator, IqOscillatorBase |
| from dcs.utils.qam_modulation import ModulationMethod, QamBasebandModulator, QamBasebandGenerator |
| from dcs.utils.symbols import MessageBinarySerializer, Symbols |
| from dcs.utils.spreading import create_spreading_config_class, Spreader, Despreader, SpreadedBasebandSpectrum3D |
| from typing import Dict, List |
| |
| 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 = 128 |
| |
| TxBasebandFilter = create_lp_filter_class(SAMPLE_RATE, 'TX Baseband') |
| |
| Spreading = create_spreading_config_class(SAMPLE_RATE) |
| |
| |
| @ui_create |
| class User(ConfigObject): |
| message: str = 'DCS' |
| spreading: Spreading = Spreading() |
| modulation: ModulationMethod = ModulationMethod.BPSK |
| tx_lp: TxBasebandFilter = TxBasebandFilter() |
| |
| def make_config_widget(self, parent: ttk.Widget) -> ConfigControlFrame: |
| frm = ConfigControlFrame(parent) |
| |
| frm1 = ConfigControlFrame(frm, borderwidth=1, relief='raised') |
| frm1.pack() |
| spreading_frame = self.spreading.make_config_widget(frm1) |
| spreading_frame.pack() |
| |
| frm2 = ConfigControlFrame(frm, borderwidth=1, relief='raised') |
| frm2.pack() |
| ttk.Label(frm2, text='Message:').grid(row=0, column=0) |
| self.ui_create_message(frm2).grid(row=0, column=1) |
| |
| ttk.Label(frm2, text='Modulation:').grid(row=1, column=0) |
| self.ui_create_modulation_dropdown(frm2).grid(row=1, column=1) |
| |
| ttk.Label(frm2, text='TX Baseband Low Pass Cut-Off:').grid(row=2, column=0) |
| self.tx_lp.ui_create_cutoff(frm2).grid(row=2, column=1) |
| ttk.Label(frm2, text='Hz').grid(row=2, column=2) |
| |
| return frm |
| |
| def make_title(self): |
| return f'{self.message} - {self.spreading.make_title()}' |
| |
| def get_symbols(self) -> Symbols: |
| return MessageBinarySerializer(self.message).to_symbols() |
| |
| def make_unspreaded_signal(self) -> TxBasebandGenerator: |
| return QamBasebandGenerator( |
| self.get_symbols(), |
| QamBasebandModulator( |
| symbol_rate=self.make_spreader().get_symbol_rate(), |
| method=ModulationMethod.BPSK, |
| baseband_filter=NoFilter(), |
| ) |
| ) |
| |
| def make_spreader(self) -> Spreader: |
| return self.spreading.make_spreader(self.modulation, self.tx_lp) |
| |
| def make_despreader(self) -> Despreader: |
| return self.spreading.make_despreader(self.modulation, self.tx_lp) |
| |
| |
| @ui_create |
| class ConfigCh07Spreading(ConfigObject): |
| _KEY = 'ch07_spreading' |
| |
| users: List[User] = [ |
| User() |
| ] |
| #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 _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 make_config_widget(self, parent: ttk.Widget) -> ConfigControlFrame: |
| frm = ConfigControlFrame(parent, borderwidth=1, relief='raised') |
| |
| user_frm = ttk.Frame(frm, borderwidth=1, relief='raised') |
| user_frm.pack() |
| ttk.Label(user_frm, text='TX Oscillator:').pack() |
| w = self.ui_create_users_list(user_frm, lambda e: e.make_title()) |
| frm.add_widget(w) |
| w.pack() |
| |
| noise_frm = self._noise_frame(frm) |
| noise_frm.pack() |
| for w in noise_frm.ctrl_widgets: |
| frm.add_widget(w) |
| |
| return frm |
| |
| |
| class Ch07SpreadingFrame(BaseFrame): |
| def __init__(self, *args, **kwargs): |
| super().__init__(*args, **kwargs) |
| |
| self._config: ConfigCh07Spreading = default_store().get_config(ConfigCh07Spreading) |
| |
| 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) |
| 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) |
| |
| fd_frm = ttk.Frame(tabs) |
| tabs.add(fd_frm, text='Frequency Domain') |
| self._fd_fig = Figure(figsize=(12, 6), dpi=100) |
| self._fd_canvas = FigureCanvasTkAgg(self._fd_fig, fd_frm) |
| self._fd_canvas.get_tk_widget().pack(expand=True, fill=BOTH) |
| fd_tb = NavigationToolbar2Tk(self._fd_canvas, fd_frm, pack_toolbar=False) |
| fd_tb.pack(side=BOTTOM) |
| |
| rx_frm = ttk.Frame(tabs) |
| tabs.add(rx_frm, text='RX Cross-Correlation') |
| 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): |
| if len(self._config.users) > 0: |
| user0 = self._config.users[0] |
| user0_spreader = user0.make_spreader() |
| tx_gen0 = user0_spreader.make_tx_baseband_generator(user0.get_symbols()) |
| |
| unspreaded_gen0 = user0.make_unspreaded_signal() |
| |
| tx_gens = copy.deepcopy(tx_gen0) |
| for user in self._config.users[1:]: |
| tx_gens += user.make_spreader().make_tx_baseband_generator(user.get_symbols()) |
| |
| class _Osc(IqOscillatorBase): |
| def make_config_widget(self, parent: ttk.Widget) -> ConfigControlFrame: |
| pass # Won't be called |
| |
| rf_gen = IqRfBandGenerator( |
| tx_gens, |
| IqUpMixer( |
| tx_osc=_Osc(freq=0), |
| noise_dBc=self._config.noise_dBc, |
| random_seed=self._config.random_seed, |
| ) |
| ) |
| # rx_gen = rf_gen.make_iq_down_mixer( |
| # IqDownMixer( |
| # rx_osc=_Osc(freq=0), |
| # baseband_filter=NoFilter(), |
| # ) |
| # ) |
| |
| rx_gen = user0.make_despreader().make_rx_correlation_generator(rf_gen) |
| |
| # Draw |
| self.draw_tx(tx_gen0, unspreaded_gen0, rf_gen) |
| self.draw_fd(rf_gen, user0_spreader) |
| self.draw_rx(rx_gen, unspreaded_gen0) |
| |
| def draw_tx(self, tx_gen0: TxBasebandGenerator, unspreaded_gen0: TxBasebandGenerator, rf_gen: RfBandGenerator): |
| self._tx_fig.clear() |
| |
| ax_msg = self._tx_fig.add_subplot(3, 1, 1) |
| ax_msg.set_xlabel('time') |
| ax_msg.set_ylabel('value') |
| ax_msg.set_title('User 0 Message (Binary)') |
| ax_u0 = self._tx_fig.add_subplot(3, 1, 2) |
| ax_u0.set_xlabel('time') |
| ax_u0.set_ylabel('value') |
| ax_u0.set_title('User 0 TX Baseband Signal') |
| ax_tx = self._tx_fig.add_subplot(3, 1, 3) |
| ax_tx.set_xlabel('time') |
| ax_tx.set_ylabel('value') |
| ax_tx.set_title('TX Baseband Signal') |
| |
| sig_msg = unspreaded_gen0.generate_tx_baseband_signal(SAMPLE_RATE) |
| ax_msg.set_xlim(np.min(sig_msg.t), np.max(sig_msg.t)) |
| ax_msg.plot(sig_msg.t, np.real(sig_msg.signal), label='Message', color='blue', linewidth=1) |
| |
| sig_u0 = tx_gen0.generate_tx_baseband_signal(SAMPLE_RATE) |
| ax_u0.set_xlim(np.min(sig_u0.t), np.max(sig_u0.t)) |
| ax_u0.plot(sig_u0.t, np.real(sig_u0.signal), label='User 0', color='red', linewidth=1) |
| |
| sig_tx = rf_gen.generate_rf_signal(SAMPLE_RATE) |
| ax_tx.set_xlim(np.min(sig_tx.t), np.max(sig_tx.t)) |
| ax_tx.plot(sig_tx.t, np.real(sig_tx.signal), label='Baseband', color='brown', linewidth=1) |
| |
| self._tx_fig.tight_layout() |
| self._tx_canvas.draw() |
| |
| def draw_fd(self, rf_gen: RfBandGenerator, spreader: Spreader): |
| self._fd_fig.clear() |
| |
| ax_fd = self._fd_fig.add_subplot(1, 1, 1, projection='3d') |
| ax_fd.set_xlabel('frequency') |
| ax_fd.set_ylabel('symbol') |
| ax_fd.set_zlabel('value (logarithmic)') |
| ax_fd.set_title('HF Signal Frequency-Domain') |
| ax_fd.set_zlim(-120, 40) |
| |
| x, y, z = SpreadedBasebandSpectrum3D( |
| rf_gen, |
| spreader.get_symbol_rate(), |
| FFT_SAMPLES_PER_SYM |
| ).make_plot_data() |
| |
| ax_fd.plot_trisurf(x, y, z) |
| |
| self._fd_fig.tight_layout() |
| self._fd_canvas.draw() |
| |
| def draw_rx(self, rx_correlation: RxCorrelationGenerator, unspreaded_gen0: TxBasebandGenerator): |
| self._rx_fig.clear() |
| |
| ax_msg = self._rx_fig.add_subplot(2, 1, 1) |
| ax_msg.set_xlabel('time') |
| ax_msg.set_ylabel('value') |
| ax_msg.set_title('User 0 Message (Binary)') |
| ax_correlation = self._rx_fig.add_subplot(2, 1, 2) |
| ax_correlation.set_xlabel('time') |
| ax_correlation.set_ylabel('value') |
| ax_correlation.set_title('RX Correlation') |
| |
| sig_msg = unspreaded_gen0.generate_tx_baseband_signal(SAMPLE_RATE) |
| ax_msg.set_xlim(np.min(sig_msg.t), np.max(sig_msg.t)) |
| ax_msg.plot(sig_msg.t, np.real(sig_msg.signal), label='Message', color='blue', linewidth=1) |
| |
| corr = rx_correlation.generate_rx_correlation(SAMPLE_RATE) |
| ax_correlation.set_xlim(np.min(corr.t), np.max(corr.t)) |
| ax_correlation.plot(corr.t, np.real(corr.signal), label='Correlation', color='blue', linewidth=1) |
| |
| self._rx_fig.tight_layout() |
| self._rx_canvas.draw() |
| |
| |
| class Ch07SpreadingWindow(Window): |
| GROUP = Ch07Group |
| TITLE = 'Spread Spectrum' |
| FRAME = Ch07SpreadingFrame |
| |
| |
| if __name__ == '__main__': |
| Ch07SpreadingWindow.main() |
| |