dcs/frames/ch06_down_sampling.py

#  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 scipy.interpolate import make_interp_spline
from dcs.frames.base import BaseFrame, Window
from dcs.frames.groups import Ch06Group
from dcs.utils import swap_freq, abs_log_fft
from typing import List, Tuple

import matplotlib
matplotlib.use('TkAgg')
from matplotlib.figure import Figure
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg, NavigationToolbar2Tk


OVERSAMPLING_LEN = 512
FFT_LEN = 2048


@ui_create
class Function(ConfigObject):
    freq: confloat(ge=0, lt=OVERSAMPLING_LEN/2, multiple_of=(4.0/OVERSAMPLING_LEN)) = 1.0
    amplitude: confloat(ge=0.0, lt=10.0, multiple_of=0.01) = 5.0
    phase: confloat(ge=-180.0, le=180.0, multiple_of=0.1) = 0.0
    offset: confloat(ge=-5.0, lt=5.0, multiple_of=0.01) = 0.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='Amplitude:').grid(row=1, column=0)
        w = self.ui_create_amplitude(frm)
        frm.add_widget(w)
        w.grid(row=1, column=1)

        ttk.Label(frm, text='Phase:').grid(row=2, column=0)
        w = self.ui_create_phase(frm)
        frm.add_widget(w)
        w.grid(row=2, column=1)
        ttk.Label(frm, text='°').grid(row=2, column=2)

        ttk.Label(frm, text='Offset:').grid(row=3, column=0)
        w = self.ui_create_offset(frm)
        frm.add_widget(w)
        w.grid(row=3, column=1)

        return frm

    def calc_signal(self, t: np.ndarray) -> np.ndarray:
        phasor = self.amplitude * np.exp(1j * self.phase * np.pi / 180)
        phi = np.exp(1j * 2 * np.pi * self.freq * t)
        return np.real(self.offset + (phasor * phi))

    def make_title(self):
        return f'f={self.freq}, {self.amplitude}, {self.phase}°'


@ui_create
class ConfigCh06DownSampling(ConfigObject):
    _KEY = 'ch06_down_sampling'

    input_funcs: List[Function] = [
        Function(freq=1.0, amplitude_dB=0.0, phase=0.0),
        Function(freq=1.5, amplitude_dB=0.0, phase=90.0),
    ]
    sample_rate: confloat(ge=0, lt=OVERSAMPLING_LEN/2, multiple_of=(4.0/OVERSAMPLING_LEN)) = 16.0
    decimation: conint(ge=0, lt=64) = 4
    lp_cutoff_freq: confloat(ge=0, lt=OVERSAMPLING_LEN/2, multiple_of=(4.0/OVERSAMPLING_LEN)) = 4.0

    def make_config_widget(self, parent: ttk.Widget) -> ConfigControlFrame:
        frm = ConfigControlFrame(parent, borderwidth=1, relief='raised')

        ttk.Label(frm, text='Input Functions:').pack()
        w = self.ui_create_input_funcs_list(frm, lambda e: e.make_title())
        frm.add_widget(w)
        w.pack()

        frm1 = ttk.Frame(frm, borderwidth=1, relief='raised')
        frm1.pack()

        ttk.Label(frm1, text='Sample Rate:').grid(row=0, column=0)
        w = self.ui_create_sample_rate(frm1)
        frm.add_widget(w)
        w.grid(row=0, column=1)

        ttk.Label(frm1, text='Decimation Factor:').grid(row=1, column=0)
        w = self.ui_create_decimation(frm1)
        frm.add_widget(w)
        w.grid(row=1, column=1)

        ttk.Label(frm1, text='Anti-Aliasing Low Pass Cut-off:').grid(row=2, column=0)
        ttk.Label(frm1, text='(0 = disable):').grid(row=3, column=1)
        w = self.ui_create_lp_cutoff_freq(frm1)
        frm.add_widget(w)
        w.grid(row=2, column=1)

        return frm

    def have_filter(self) -> bool:
        return not (self.lp_cutoff_freq == 0)

    def _decimate_vec(self, vec: np.ndarray) -> np.ndarray:
        return np.array([vec[idx] for idx in range(0, len(vec), self.decimation)])

    def calc_input_signal(self, t: np.ndarray) -> np.ndarray:
        x = np.zeros((len(self.input_funcs), len(t)), dtype=np.float64)
        for index, func in enumerate(self.input_funcs):
            x[index, :] = func.calc_signal(t)
        return np.sum(x, axis=0)

    def calc_decimated_bypassed_signal(self, t: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
        sig = self.calc_input_signal(t)
        return self._decimate_vec(t), self._decimate_vec(sig)

    def calc_filtered_signal(self, t: np.ndarray) -> np.ndarray:
        inp = self.calc_input_signal(t)

        if self.have_filter():
            b, a = scipy.signal.cheby1(N=5, Wn=self.lp_cutoff_freq, rp=1, btype='low', fs=self.sample_rate)
            zi = scipy.signal.lfilter_zi(b, a)
            z, _ = scipy.signal.lfilter(b, a, inp, zi=zi*inp[0])
            return z
        else:
            return inp

    def calc_decimated_signal(self, t: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
        sig = self.calc_filtered_signal(t)
        return self._decimate_vec(t), self._decimate_vec(sig)


class Ch06DownSamplingFrame(BaseFrame):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self._config: ConfigCh06DownSampling = default_store().get_config(ConfigCh06DownSampling)

        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.draw_td()
        self.draw_fd()

    def _create_signal_tabs(self) -> ttk.Widget:
        tabs = ttk.Notebook(self)

        td_frm = ttk.Frame(tabs)
        tabs.add(td_frm, text='Time Domain')
        self._td_fig = Figure(figsize=(12, 6), dpi=100)
        self._td_canvas = FigureCanvasTkAgg(self._td_fig, td_frm)
        self._td_canvas.get_tk_widget().pack(expand=True, fill=BOTH)
        td_tb = NavigationToolbar2Tk(self._td_canvas, td_frm, pack_toolbar=False)
        td_tb.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)

        self.draw_td()
        self.draw_fd()

        return tabs

    def draw_td(self):
        self._td_fig.clear()

        if self._config.have_filter():
            ax_inp = self._td_fig.add_subplot(3, 1, 1)
        else:
            ax_inp = self._td_fig.add_subplot(2, 1, 1)
        ax_inp.set_xlim(0.0, 1.1)
        ax_inp.set_xlabel('time')
        ax_inp.set_ylabel('value')
        ax_inp.set_title('Input Signal')
        if self._config.have_filter():
            ax_flt = self._td_fig.add_subplot(3, 1, 2)
            ax_flt.set_xlim(0.0, 1.1)
            ax_flt.set_xlabel('time')
            ax_flt.set_ylabel('value')
            ax_flt.set_title('Signal After Anti-Aliasing Filter')
            ax_dec = self._td_fig.add_subplot(3, 1, 3)
        else:
            ax_dec = self._td_fig.add_subplot(2, 1, 2)
        ax_dec.set_xlim(0.0, 1.1)
        ax_dec.set_xlabel('time')
        ax_dec.set_ylabel('value')
        ax_dec.set_title('Decimated Signal')

        t = np.arange(0, 1.1, 1.0/self._config.sample_rate)
        if len(t) > 3:
            t_interp = np.linspace(np.min(t), np.max(t), OVERSAMPLING_LEN)
        else:
            t_interp = np.zeros((0,))

        x_inp = self._config.calc_input_signal(t)
        t_inp_dec, x_inp_dec = self._config.calc_decimated_bypassed_signal(t)
        if len(t) > 3:
            fn_inp_interp = make_interp_spline(t, x_inp)
            x_inp_interp = fn_inp_interp(t_interp)
        else:
            x_inp_interp = np.zeros((0,))

        if self._config.have_filter():
            x_flt = self._config.calc_filtered_signal(t)
            if len(t) > 3:
                fn_flt_interp = make_interp_spline(t, x_flt)
                x_flt_interp = fn_flt_interp(t_interp)
            else:
                x_flt_interp = np.zeros((0,))

        t_dec, x_dec = self._config.calc_decimated_signal(t)
        if len(t) > 3:
            fn_dec_interp = make_interp_spline(t_dec, x_dec)
            x_dec_interp = fn_dec_interp(t_interp)
        else:
            x_dec_interp = np.zeros((0,))

        ax_inp.plot(t_inp_dec, x_inp_dec, label='Input Signal (Decimated)', marker='o', linestyle='none', color='red', linewidth=1)
        ax_inp.plot(t, x_inp, label='Input Signal (Sampled)', marker='x', linestyle='none', color='blue', linewidth=1)
        ax_inp.plot(t_interp, x_inp_interp, label='Input Signal (Interpolated)', linestyle='dashed', color='blue', linewidth=1)
        ax_inp.legend()

        if self._config.have_filter():
            ax_flt.plot(t_dec, x_dec, label='Filtered Signal (Decimated)', marker='o', linestyle='none', color='red', linewidth=1)
            ax_flt.plot(t, x_flt, label='Filtered Signal (Sampled)', marker='x', linestyle='none', color='green', linewidth=1)
            ax_flt.plot(t_interp, x_flt_interp, label='Filtered Signal (Interpolated)', linestyle='dashed', color='green', linewidth=1)
            ax_flt.legend()

        ax_dec.plot(t_dec, x_dec, label='Decimated Signal (Sampled)', marker='x', linestyle='none', color='red', linewidth=1)
        ax_dec.plot(t_interp, x_dec_interp, label='Decimated Signal (Interpolated)', linestyle='dashed', color='red', linewidth=1)
        ax_dec.legend()

        self._td_fig.tight_layout()
        self._td_canvas.draw()

    def draw_fd(self):
        self._fd_fig.clear()

        if self._config.have_filter():
            ax_inp = self._fd_fig.add_subplot(3, 1, 1)
        else:
            ax_inp = self._fd_fig.add_subplot(2, 1, 1)
        ax_inp.set_xlim(-self._config.sample_rate/2, self._config.sample_rate/2)
        ax_inp.set_xlabel('frequency')
        ax_inp.set_ylabel('value (dB)')
        ax_inp.set_title('Input Signal')
        if self._config.have_filter():
            ax_flt = self._fd_fig.add_subplot(3, 1, 2)
            ax_flt.set_xlim(-self._config.sample_rate/2, self._config.sample_rate/2)
            ax_flt.set_xlabel('frequency')
            ax_flt.set_ylabel('value (dB)')
            ax_flt.set_title('Signal After Anti-Aliasing Filter')
            ax_dec = self._fd_fig.add_subplot(3, 1, 3)
        else:
            ax_dec = self._fd_fig.add_subplot(2, 1, 2)
        ax_dec.set_xlim(-self._config.sample_rate/(self._config.decimation * 2), self._config.sample_rate/(self._config.decimation * 2))
        ax_dec.set_xlabel('frequency')
        ax_dec.set_ylabel('value (dB)')
        ax_dec.set_title('Decimated Signal')

        t = np.arange(0, FFT_LEN, 1) / self._config.sample_rate
        f = swap_freq(np.fft.fftfreq(t.shape[-1], 1.0/self._config.sample_rate))

        x_inp = self._config.calc_input_signal(t)

        if self._config.have_filter():
            x_flt = self._config.calc_filtered_signal(t)

        t_dec, x_dec = self._config.calc_decimated_signal(t)
        f_dec = swap_freq(np.fft.fftfreq(t_dec.shape[-1], 1.0 * self._config.decimation / self._config.sample_rate))

        ax_inp.plot(f, abs_log_fft(x_inp), label='Input Signal', linestyle='solid', color='blue', linewidth=1)
        ax_inp.legend()

        if self._config.have_filter():
            ax_flt.plot(f, abs_log_fft(x_flt), label='Filtered Signal', linestyle='solid', color='green', linewidth=1)
            ax_flt.legend()

        ax_dec.plot(f_dec, abs_log_fft(x_dec), label='Decimated Signal', linestyle='solid', color='red', linewidth=1)
        ax_dec.legend()

        self._fd_fig.tight_layout()
        self._fd_canvas.draw()


class Ch06DownSamplingWindow(Window):
    GROUP = Ch06Group
    TITLE = 'Down Sampling'
    FRAME = Ch06DownSamplingFrame


if __name__ == '__main__':
    Ch06DownSamplingWindow.main()