feat: IQ Mixer Utility
Signed-off-by: Philipp Le <philipp@philipple.de>
Change-Id: I11377f2361564d399fbadf9d22b4735a81015d79
diff --git a/dcs/utils/iq_mixer.py b/dcs/utils/iq_mixer.py
new file mode 100644
index 0000000..6e1ae23
--- /dev/null
+++ b/dcs/utils/iq_mixer.py
@@ -0,0 +1,160 @@
+# 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
+from pydantic import confloat
+from dcs.config import ConfigObject, ui_create, ConfigControlFrame
+from typing import Dict
+import numpy as np
+from abc import abstractmethod, ABC
+
+from .signal import Signal
+from .chain import TxBasebandGenerator, RfBandGenerator, RxBasebandGenerator
+from .filter import Filter
+
+
+@ui_create
+class IqOscillatorBase(ConfigObject, ABC):
+ freq: confloat(ge=0) = 100.0
+ phase: confloat(ge=-180.0, le=180.0, multiple_of=0.1) = 0.0
+ phase_imbalance: confloat(ge=-90.0, le=90.0, multiple_of=0.01) = 0.0
+ amplitude_imbalance: confloat(ge=-1.0, le=1.0, multiple_of=0.001) = 0.0
+
+ def calc_signal(self, t: np.ndarray) -> np.ndarray:
+ phase_rad = self.phase * np.pi / 180
+ phi = 2 * np.pi * self.freq * t
+ out_i = np.cos(phi + phase_rad)
+ out_q = (1 + self.amplitude_imbalance) * np.sin(phi + phase_rad + self.phase_imbalance)
+ return out_i + (1j * out_q)
+
+ @abstractmethod
+ def make_config_widget(self, parent: ttk.Widget) -> ConfigControlFrame:
+ ...
+
+ def make_title(self):
+ return f'f={self.freq}, phi={self.phase}°'
+
+
+def create_iq_oscillator_class(sample_rate: int, imbalance_wdg: bool = True):
+ @ui_create
+ class _IqOscillator(IqOscillatorBase):
+ freq: confloat(ge=0, lt=sample_rate/4, multiple_of=(4.0/sample_rate)) = 100.0
+
+ def make_config_widget(self, parent: ttk.Widget) -> ConfigControlFrame:
+ frm = ConfigControlFrame(parent)
+
+ ttk.Label(frm, text='Frequency:').grid(row=0, column=0)
+ w = self.ui_create_freq(frm)
+ frm.add_widget(w)
+ w.grid(row=0, column=1)
+
+ ttk.Label(frm, text='Phase:').grid(row=1, column=0)
+ w = self.ui_create_phase(frm)
+ frm.add_widget(w)
+ w.grid(row=1, column=1)
+ ttk.Label(frm, text='°').grid(row=1, column=2)
+
+ if imbalance_wdg:
+ ttk.Label(frm, text='Phase Imbalance:').grid(row=2, column=0)
+ w = self.ui_create_phase_imbalance(frm)
+ frm.add_widget(w)
+ w.grid(row=2, column=1)
+ ttk.Label(frm, text='°').grid(row=2, column=2)
+
+ ttk.Label(frm, text='Amplitude Imbalance:').grid(row=3, column=0)
+ w = self.ui_create_amplitude_imbalance(frm)
+ frm.add_widget(w)
+ w.grid(row=3, column=1)
+ ttk.Label(frm, text='Rel. Error').grid(row=3, column=2)
+
+ return frm
+
+ _IqOscillator.update_forward_refs(sample_rate=sample_rate)
+ return _IqOscillator
+
+
+class IqUpMixer:
+ def __init__(self, tx_osc: IqOscillatorBase, noise_dBc: float, random_seed: int):
+ self.tx_osc = tx_osc
+ self.noise_dBc = noise_dBc
+ self.random_seed = random_seed
+
+ def _make_noise(self, t: np.ndarray) -> np.ndarray:
+ np.random.seed(self.random_seed)
+ noise_rms = np.power(10, (self.noise_dBc / 20))
+ noise_sigma = np.sqrt(2) * noise_rms
+ return np.random.normal(0, noise_sigma, len(t))
+
+ def calc_rf_signal(self, tx_baseband_signal: Signal) -> Signal:
+ t = tx_baseband_signal.t
+ baseband = tx_baseband_signal.signal
+ osc = self.tx_osc.calc_signal(t)
+ i_mixed = np.real(baseband) * np.real(osc)
+ q_mixed = np.imag(baseband) * np.imag(osc)
+ return Signal(
+ t=t,
+ signal=i_mixed - q_mixed + self._make_noise(t),
+ sample_rate=tx_baseband_signal.sample_rate,
+ )
+
+
+class IqDownMixer:
+ def __init__(self, rx_osc: IqOscillatorBase, baseband_filter: Filter):
+ self.rx_osc = rx_osc
+ self.baseband_filter = baseband_filter
+
+ def calc_rx_baseband_signal(self, rf_signal: Signal) -> Signal:
+ osc = self.rx_osc.calc_signal(rf_signal.t)
+ i_mixed = 2 * rf_signal.signal * np.real(osc)
+ q_mixed = 2 * rf_signal.signal * np.imag(osc)
+ base = i_mixed - (1j * q_mixed)
+ return self.baseband_filter.filter(
+ Signal(
+ t=rf_signal.t,
+ signal=base,
+ sample_rate=rf_signal.sample_rate,
+ )
+ )
+
+
+class IqTxBasebandGenerator(TxBasebandGenerator, ABC):
+ @abstractmethod
+ def generate_tx_baseband_signal(self, sample_rate: float) -> Signal:
+ ...
+
+ def make_rf_band_generator(self, up_mix: IqUpMixer) -> IqRfBandGenerator:
+ return IqRfBandGenerator(self, up_mix)
+
+
+class IqRfBandGenerator(RfBandGenerator):
+ def __init__(self, tx_baseband_generator: IqTxBasebandGenerator, up_mix: IqUpMixer):
+ self.tx_baseband_generator = tx_baseband_generator
+ self.up_mix = up_mix
+ self._lazy_eval_store: Dict[float, Signal] = {}
+
+ def generate_rf_signal(self, sample_rate: float) -> Signal:
+ if sample_rate not in self._lazy_eval_store:
+ tx_baseband_signal = self.tx_baseband_generator.generate_tx_baseband_signal(sample_rate)
+ self._lazy_eval_store[sample_rate] = self.up_mix.calc_rf_signal(tx_baseband_signal)
+ return self._lazy_eval_store[sample_rate]
+
+ def make_iq_down_mixer(self, down_mix: IqDownMixer) -> IqRxBasebandGenerator:
+ return IqRxBasebandGenerator(self, down_mix)
+
+
+class IqRxBasebandGenerator(RxBasebandGenerator):
+ def __init__(self, rf_generator: IqRfBandGenerator, down_mix: IqDownMixer):
+ self.rf_generator = rf_generator
+ self.down_mix = down_mix
+ self._lazy_eval_store: Dict[float, Signal] = {}
+
+ def generate_rx_baseband_signal(self, sample_rate: float) -> Signal:
+ if sample_rate not in self._lazy_eval_store:
+ rf_signal = self.rf_generator.generate_rf_signal(sample_rate)
+ self._lazy_eval_store[sample_rate] = self.down_mix.calc_rx_baseband_signal(rf_signal)
+ return self._lazy_eval_store[sample_rate]