#!/usr/bin/env python
# Copyright 2012,2013 Free Software Foundation, Inc.
#
# This file is part of GNU Radio
#
# SPDX-License-Identifier: GPL-3.0-or-later
#
#


import numpy

from gnuradio import gr, gr_unittest, digital, blocks
import pmt


class qa_ofdm_frame_equalizer_vcvc (gr_unittest.TestCase):

    def setUp(self):
        self.tb = gr.top_block()
        self.tsb_key = "tsb_key"

    def tearDown(self):
        self.tb = None

    def test_001_simple(self):
        """ Very simple functionality testing:
        - static equalizer
        - init channel state with all ones
        - transmit all ones
        - make sure we rx all ones
        - Tag check: put in frame length tag and one other random tag,
                     make sure they're propagated
        """
        fft_len = 8
        equalizer = digital.ofdm_equalizer_static(fft_len)
        n_syms = 3
        tx_data = [1, ] * fft_len * n_syms
        chan_tag = gr.tag_t()
        chan_tag.offset = 0
        chan_tag.key = pmt.string_to_symbol("ofdm_sync_chan_taps")
        chan_tag.value = pmt.init_c32vector(fft_len, (1,) * fft_len)
        random_tag = gr.tag_t()
        random_tag.offset = 1
        random_tag.key = pmt.string_to_symbol("foo")
        random_tag.value = pmt.from_long(42)
        src = blocks.vector_source_c(
            tx_data, False, fft_len, (chan_tag, random_tag))
        eq = digital.ofdm_frame_equalizer_vcvc(
            equalizer.base(), 0, self.tsb_key)
        sink = blocks.tsb_vector_sink_c(fft_len, tsb_key=self.tsb_key)
        self.tb.connect(
            src,
            blocks.stream_to_tagged_stream(
                gr.sizeof_gr_complex,
                fft_len,
                n_syms,
                self.tsb_key),
            eq,
            sink)
        self.tb.run()
        # Check data
        self.assertEqual(tx_data, sink.data()[0])
        # Check tags
        tag_dict = dict()
        for tag in sink.tags():
            ptag = gr.tag_to_python(tag)
            tag_dict[ptag.key] = ptag.value
        expected_dict = {
            'foo': 42
        }
        self.assertEqual(tag_dict, expected_dict)

    def test_001b_simple_skip_nothing(self):
        """
        Same as before, but put a skip-header in there
        """
        fft_len = 8
        equalizer = digital.ofdm_equalizer_static(fft_len, symbols_skipped=1)
        n_syms = 3
        tx_data = [1, ] * fft_len * n_syms
        chan_tag = gr.tag_t()
        chan_tag.offset = 0
        chan_tag.key = pmt.string_to_symbol("ofdm_sync_chan_taps")
        chan_tag.value = pmt.init_c32vector(fft_len, (1,) * fft_len)
        src = blocks.vector_source_c(tx_data, False, fft_len, (chan_tag,))
        eq = digital.ofdm_frame_equalizer_vcvc(
            equalizer.base(), 0, self.tsb_key)
        sink = blocks.tsb_vector_sink_c(fft_len, tsb_key=self.tsb_key)
        self.tb.connect(
            src,
            blocks.stream_to_tagged_stream(
                gr.sizeof_gr_complex,
                fft_len,
                n_syms,
                self.tsb_key),
            eq,
            sink)
        self.tb.run()
        # Check data
        self.assertEqual(tx_data, sink.data()[0])

    def test_001c_carrier_offset_no_cp(self):
        """
        Same as before, but put a carrier offset in there
        """
        fft_len = 8
        cp_len = 0
        n_syms = 1
        carr_offset = 1
        occupied_carriers = ((-2, -1, 1, 2),)
        tx_data = (
            0, 0, 0, -1j, -1j, 0, -1j, -1j,
        )
        # The rx'd signal is shifted
        rx_expected = (0, 0, 1, 1, 0, 1, 1, 0) * n_syms
        equalizer = digital.ofdm_equalizer_static(fft_len, occupied_carriers)
        chan_tag = gr.tag_t()
        chan_tag.offset = 0
        chan_tag.key = pmt.string_to_symbol("ofdm_sync_chan_taps")
        # Note: this is shifted to the correct position!
        chan_tag.value = pmt.init_c32vector(
            fft_len, (0, 0, -1j, -1j, 0, -1j, -1j, 0))
        offset_tag = gr.tag_t()
        offset_tag.offset = 0
        offset_tag.key = pmt.string_to_symbol("ofdm_sync_carr_offset")
        offset_tag.value = pmt.from_long(carr_offset)
        src = blocks.vector_source_c(
            tx_data, False, fft_len, (chan_tag, offset_tag))
        eq = digital.ofdm_frame_equalizer_vcvc(
            equalizer.base(), cp_len, self.tsb_key)
        sink = blocks.tsb_vector_sink_c(fft_len, tsb_key=self.tsb_key)
        self.tb.connect(
            src,
            blocks.stream_to_tagged_stream(
                gr.sizeof_gr_complex,
                fft_len,
                n_syms,
                self.tsb_key),
            eq,
            sink)
        self.tb.run()
        # Check data
        self.assertComplexTuplesAlmostEqual(
            rx_expected, sink.data()[0], places=4)

    def test_001c_carrier_offset_cp(self):
        """
        Same as before, but put a carrier offset in there and a CP
        """
        fft_len = 8
        cp_len = 2
        n_syms = 3
        # cp_len/fft_len == 1/4, therefore, the phase is rotated by
        # carr_offset * \pi/2 in every symbol
        occupied_carriers = ((-2, -1, 1, 2),)
        carr_offset = -1
        tx_data = (
            0, -1j, -1j, 0, -1j, -1j, 0, 0,
            0, -1, -1, 0, -1, -1, 0, 0,
            0, 1j, 1j, 0, 1j, 1j, 0, 0,
        )
        # Rx'd signal is corrected
        rx_expected = (0, 0, 1, 1, 0, 1, 1, 0) * n_syms
        equalizer = digital.ofdm_equalizer_static(fft_len, occupied_carriers)
        chan_tag = gr.tag_t()
        chan_tag.offset = 0
        chan_tag.key = pmt.string_to_symbol("ofdm_sync_chan_taps")
        chan_tag.value = pmt.init_c32vector(fft_len, (0, 0, 1, 1, 0, 1, 1, 0))
        offset_tag = gr.tag_t()
        offset_tag.offset = 0
        offset_tag.key = pmt.string_to_symbol("ofdm_sync_carr_offset")
        offset_tag.value = pmt.from_long(carr_offset)
        src = blocks.vector_source_c(
            tx_data, False, fft_len, (chan_tag, offset_tag))
        eq = digital.ofdm_frame_equalizer_vcvc(
            equalizer.base(), cp_len, self.tsb_key)
        sink = blocks.tsb_vector_sink_c(fft_len, tsb_key=self.tsb_key)
        self.tb.connect(
            src,
            blocks.stream_to_tagged_stream(
                gr.sizeof_gr_complex,
                fft_len,
                n_syms,
                self.tsb_key),
            eq,
            sink)
        self.tb.run()
        # Check data
        self.assertComplexTuplesAlmostEqual(
            rx_expected, sink.data()[0], places=4)

    def test_002_static(self):
        """
        - Add a simple channel
        - Make symbols QPSK
        """
        fft_len = 8
        #           4   5  6  7   0  1  2   3
        tx_data = [-1, -1, 1, 2, -1, 3, 0, -1,  # 0
                   -1, -1, 0, 2, -1, 2, 0, -1,  # 8
                   -1, -1, 3, 0, -1, 1, 0, -1,  # 16 (Pilot symbols)
                   -1, -1, 1, 1, -1, 0, 2, -1]  # 24
        cnst = digital.constellation_qpsk()
        tx_signal = [
            cnst.map_to_points_v(x)[0] if x != -
            1 else 0 for x in tx_data]
        occupied_carriers = ((1, 2, 6, 7),)
        pilot_carriers = ((), (), (1, 2, 6, 7), ())
        pilot_symbols = (
            [], [], [cnst.map_to_points_v(x)[0] for x in (1, 0, 3, 0)], []
        )
        equalizer = digital.ofdm_equalizer_static(
            fft_len, occupied_carriers, pilot_carriers, pilot_symbols)
        channel = [
            0, 0, 1, 1, 0, 1, 1, 0,
            # These coefficients will be rotated slightly (but less than \pi/2)
            0, 0, 1, 1, 0, 1, 1, 0,
            0, 0, 1j, 1j, 0, 1j, 1j, 0,  # Go crazy here!
            0, 0, 1j, 1j, 0, 1j, 1j, 0
        ]
        channel = [
            0, 0, 1, 1, 0, 1, 1, 0,
            # These coefficients will be rotated slightly (but less than \pi/2)
            0, 0, 1, 1, 0, 1, 1, 0,
            0, 0, 1j, 1j, 0, 1j, 1j, 0,  # Go crazy here!
            0, 0, 1j, 1j, 0, 1j, 1j, 0
        ]
        for idx in range(fft_len, 2 * fft_len):
            channel[idx] = channel[idx - fft_len] * \
                numpy.exp(1j * .1 * numpy.pi * (numpy.random.rand() - .5))
        chan_tag = gr.tag_t()
        chan_tag.offset = 0
        chan_tag.key = pmt.string_to_symbol("ofdm_sync_chan_taps")
        chan_tag.value = pmt.init_c32vector(fft_len, channel[:fft_len])
        src = blocks.vector_source_c(numpy.multiply(
            tx_signal, channel), False, fft_len, (chan_tag,))
        sink = blocks.tsb_vector_sink_c(vlen=fft_len, tsb_key=self.tsb_key)
        eq = digital.ofdm_frame_equalizer_vcvc(
            equalizer.base(), 0, self.tsb_key, True)
        self.tb.connect(
            src,
            blocks.stream_to_tagged_stream(
                gr.sizeof_gr_complex,
                fft_len,
                len(tx_data) //
                fft_len,
                self.tsb_key),
            eq,
            sink)
        self.tb.run()
        rx_data = [
            cnst.decision_maker_v(
                (x,)) if x != 0 else -1 for x in sink.data()[0]]
        # Check data
        self.assertEqual(tx_data, rx_data)
        # Check tags
        tag_dict = dict()
        for tag in sink.tags():
            ptag = gr.tag_to_python(tag)
            tag_dict[ptag.key] = ptag.value
            if ptag.key == 'ofdm_sync_chan_taps':
                tag_dict[ptag.key] = list(pmt.c32vector_elements(tag.value))
            else:
                tag_dict[ptag.key] = pmt.to_python(tag.value)
        expected_dict = {
            'ofdm_sync_chan_taps': channel[-fft_len:]
        }
        # Won't be exactly the same when compiled with -fcx-limited-range
        self.assertTrue(
            numpy.allclose(
                tag_dict['ofdm_sync_chan_taps'],
                expected_dict['ofdm_sync_chan_taps']))
        expected_dict['ofdm_sync_chan_taps'] = tag_dict['ofdm_sync_chan_taps']
        self.assertEqual(tag_dict, expected_dict)

    def test_002_static_wo_tags(self):
        """ Same as before, but the input stream has no tag.
        We specify the frame size in the constructor.
        We also specify a tag key, so the output stream *should* have
        a TSB tag.
        """
        fft_len = 8
        n_syms = 4
        #           4   5  6  7   0  1  2   3
        tx_data = [-1, -1, 1, 2, -1, 3, 0, -1,  # 0
                   -1, -1, 0, 2, -1, 2, 0, -1,  # 8
                   -1, -1, 3, 0, -1, 1, 0, -1,  # 16 (Pilot symbols)
                   -1, -1, 1, 1, -1, 0, 2, -1]  # 24
        cnst = digital.constellation_qpsk()
        tx_signal = [
            cnst.map_to_points_v(x)[0] if x != -
            1 else 0 for x in tx_data]
        occupied_carriers = ((1, 2, 6, 7),)
        pilot_carriers = ((), (), (1, 2, 6, 7), ())
        pilot_symbols = (
            [], [], [cnst.map_to_points_v(x)[0] for x in (1, 0, 3, 0)], []
        )
        equalizer = digital.ofdm_equalizer_static(
            fft_len, occupied_carriers, pilot_carriers, pilot_symbols)
        channel = [
            0, 0, 1, 1, 0, 1, 1, 0,
            # These coefficients will be rotated slightly (below)...
            0, 0, 1, 1, 0, 1, 1, 0,
            0, 0, 1j, 1j, 0, 1j, 1j, 0,  # Go crazy here!
            0, 0, 1j, 1j, 0, 1j, 1j, 0  # ...and again here.
        ]
        for idx in range(fft_len, 2 * fft_len):
            channel[idx] = channel[idx - fft_len] * \
                numpy.exp(1j * .1 * numpy.pi * (numpy.random.rand() - .5))
            idx2 = idx + 2 * fft_len
            channel[idx2] = channel[idx2] * \
                numpy.exp(1j * 0 * numpy.pi * (numpy.random.rand() - .5))
        src = blocks.vector_source_c(
            numpy.multiply(
                tx_signal,
                channel),
            False,
            fft_len)
        eq = digital.ofdm_frame_equalizer_vcvc(
            equalizer.base(), 0, self.tsb_key, False, n_syms)
        sink = blocks.tsb_vector_sink_c(vlen=fft_len, tsb_key=self.tsb_key)
        self.tb.connect(
            src,
            blocks.stream_to_tagged_stream(
                gr.sizeof_gr_complex,
                fft_len,
                len(tx_data) //
                fft_len,
                self.tsb_key),
            eq,
            sink)
        self.tb.run()
        rx_data = [
            cnst.decision_maker_v(
                (x,)) if x != 0 else -1 for x in sink.data()[0]]
        self.assertEqual(tx_data, rx_data)
        # Check TSB Functionality
        packets = sink.data()
        self.assertEqual(len(packets), 1)
        self.assertEqual(len(packets[0]), len(tx_data))

    def test_002_static_wo_tags(self):
        fft_len = 8
        #           4   5  6  7   0  1  2   3
        tx_data = [-1, -1, 1, 2, -1, 3, 0, -1,  # 0
                   -1, -1, 0, 2, -1, 2, 0, -1,  # 8
                   -1, -1, 3, 0, -1, 1, 0, -1,  # 16 (Pilot symbols)
                   -1, -1, 1, 1, -1, 0, 2, -1]  # 24
        cnst = digital.constellation_qpsk()
        tx_signal = [
            cnst.map_to_points_v(x)[0] if x != -
            1 else 0 for x in tx_data]
        occupied_carriers = ((1, 2, 6, 7),)
        pilot_carriers = ((), (), (1, 2, 6, 7), ())
        pilot_symbols = (
            [], [], [cnst.map_to_points_v(x)[0] for x in (1, 0, 3, 0)], []
        )
        equalizer = digital.ofdm_equalizer_static(
            fft_len, occupied_carriers, pilot_carriers, pilot_symbols)
        channel = [
            0, 0, 1, 1, 0, 1, 1, 0,
            # These coefficients will be rotated slightly...
            0, 0, 1, 1, 0, 1, 1, 0,
            0, 0, 1j, 1j, 0, 1j, 1j, 0,  # Go crazy here!
            0, 0, 1j, 1j, 0, 1j, 1j, 0  # ...and again here.
        ]
        for idx in range(fft_len, 2 * fft_len):
            channel[idx] = channel[idx - fft_len] * \
                numpy.exp(1j * .1 * numpy.pi * (numpy.random.rand() - .5))
            idx2 = idx + 2 * fft_len
            channel[idx2] = channel[idx2] * \
                numpy.exp(1j * 0 * numpy.pi * (numpy.random.rand() - .5))
        src = blocks.vector_source_c(
            numpy.multiply(
                tx_signal,
                channel),
            False,
            fft_len)
        sink = blocks.vector_sink_c(fft_len)
        eq = digital.ofdm_frame_equalizer_vcvc(
            equalizer.base(), 0, "", False, 4)
        self.tb.connect(src, eq, sink)
        self.tb.run()
        rx_data = [
            cnst.decision_maker_v(
                (x,)) if x != 0 else -1 for x in sink.data()]
        self.assertEqual(tx_data, rx_data)

    def test_002_simpledfe(self):
        """ Use the simple DFE equalizer. """
        fft_len = 8
        #           4   5  6  7   0  1  2   3
        tx_data = [-1, -1, 1, 2, -1, 3, 0, -1,  # 0
                   -1, -1, 0, 2, -1, 2, 0, -1,  # 8
                   -1, -1, 3, 0, -1, 1, 0, -1,  # 16 (Pilot symbols)
                   -1, -1, 1, 1, -1, 0, 2, -1]  # 24
        cnst = digital.constellation_qpsk()
        tx_signal = [
            cnst.map_to_points_v(x)[0] if x != -
            1 else 0 for x in tx_data]
        occupied_carriers = ((1, 2, 6, 7),)
        pilot_carriers = ((), (), (1, 2, 6, 7), ())
        pilot_symbols = (
            [], [], [cnst.map_to_points_v(x)[0] for x in (1, 0, 3, 0)], []
        )
        equalizer = digital.ofdm_equalizer_simpledfe(
            fft_len,
            cnst.base(),
            occupied_carriers,
            pilot_carriers,
            pilot_symbols,
            0,
            0.01)
        equalizer_soft = digital.ofdm_equalizer_simpledfe(
            fft_len,
            cnst.base(),
            occupied_carriers,
            pilot_carriers,
            pilot_symbols,
            0,
            0.01,
            enable_soft_output=True)
        channel = [
            0, 0, 1, 1, 0, 1, 1, 0,
            # These coefficients will be rotated slightly...
            0, 0, 1, 1, 0, 1, 1, 0,
            0, 0, 1j, 1j, 0, 1j, 1j, 0,  # Go crazy here!
            0, 0, 1j, 1j, 0, 1j, 1j, 0  # ...and again here.
        ]
        for idx in range(fft_len, 2 * fft_len):
            channel[idx] = channel[idx - fft_len] * \
                numpy.exp(1j * .1 * numpy.pi * (numpy.random.rand() - .5))
            idx2 = idx + 2 * fft_len
            channel[idx2] = channel[idx2] * \
                numpy.exp(1j * 0 * numpy.pi * (numpy.random.rand() - .5))
        chan_tag = gr.tag_t()
        chan_tag.offset = 0
        chan_tag.key = pmt.string_to_symbol("ofdm_sync_chan_taps")
        chan_tag.value = pmt.init_c32vector(fft_len, channel[:fft_len])
        src = blocks.vector_source_c(numpy.multiply(
            tx_signal, channel), False, fft_len, (chan_tag,))
        eq = digital.ofdm_frame_equalizer_vcvc(
            equalizer.base(), 0, self.tsb_key, True)
        eq_soft = digital.ofdm_frame_equalizer_vcvc(
            equalizer_soft.base(), 0, self.tsb_key, True)
        sink = blocks.tsb_vector_sink_c(fft_len, tsb_key=self.tsb_key)
        sink_soft = blocks.tsb_vector_sink_c(fft_len, tsb_key=self.tsb_key)
        stream_to_tagged = blocks.stream_to_tagged_stream(
            gr.sizeof_gr_complex, fft_len, len(tx_data) // fft_len, self.tsb_key)
        self.tb.connect(
            src,
            stream_to_tagged,
            eq,
            sink
        )
        self.tb.connect(
            stream_to_tagged,
            eq_soft,
            sink_soft
        )
        self.tb.run()

        out_syms = numpy.array(sink.data()[0])
        out_syms_soft = numpy.array(sink_soft.data()[0])

        def demod(syms): return [
            cnst.decision_maker_v(
                (x,)) if x != 0 else -1 for x in syms]
        rx_data = demod(out_syms)
        rx_data_soft = demod(out_syms_soft)

        # Uncomment to plot symbols
        #import matplotlib.pyplot as plt
        #def plot_syms(d): plt.figure(); plt.plot(d.real, d.imag, 'b.')
        #
        # plot_syms(out_syms)
        # plot_syms(out_syms_soft)
        # plt.show()

        self.assertEqual(tx_data, rx_data)
        self.assertEqual(rx_data, rx_data_soft)
        self.assertFalse(numpy.allclose(out_syms, out_syms_soft))
        self.assertEqual(len(sink.tags()), 1)
        tag = sink.tags()[0]
        self.assertEqual(pmt.symbol_to_string(tag.key), "ofdm_sync_chan_taps")
        self.assertComplexTuplesAlmostEqual(
            list(pmt.c32vector_elements(tag.value)), channel[-fft_len:], places=1)


if __name__ == '__main__':
    gr_unittest.run(qa_ofdm_frame_equalizer_vcvc)
