tests.py

import unittest

import numpy as np

import vhsdecode.process as process
import vhsdecode.utils as utils
from vhsdecode.sync import calczc as c_calczc


class DemodTest(unittest.TestCase):
    def test_ire(self):
        """Check that the IRE of the output corresponds to the correct frequencies."""
        samplerate_mhz = 40
        decoder = process.VHSRFDecode(inputfreq=samplerate_mhz, system="PAL")
        max_hz = 4800000
        max_mhz = max_hz / 1000000
        min_hz = 3800000
        min_mhz = min_hz / 1000000
        wavemax = utils.gen_wave_at_frequency(
            max_mhz, samplerate_mhz, decoder.blocklen / 2
        )
        wavemin = utils.gen_wave_at_frequency(
            min_mhz, samplerate_mhz, decoder.blocklen / 2
        )
        wave = np.concatenate((wavemax, wavemin))
        demod = decoder.demodblock(data=wave)["video"]["demod"]

        # Reference white
        max_ire_abs = decoder.iretohz(100)
        sync_ire_abs = decoder.iretohz(decoder.SysParams["vsync_ire"])
        print("max: ", max_ire_abs)
        print("min_ire: ", decoder.SysParams["vsync_ire"])
        print("min: ", sync_ire_abs)

        max_demod = demod[1000 : (decoder.blocklen // 2) - 1000]
        np.testing.assert_allclose(
            max_demod, np.full(len(max_demod), max_hz), rtol=1.5e-04, atol=20
        )
        min_demod = demod[(decoder.blocklen // 2) + 1000 : decoder.blocklen - 1000]
        # Demodulated signal fluctuates more at the sync end than at the top end,
        # so allowing a little more tolerance here.
        np.testing.assert_allclose(
            min_demod, np.full(len(min_demod), min_hz), rtol=3e-04, atol=20
        )

    def test_ire_ntsc(self):
        """Check that the IRE of the output corresponds to the correct frequencies."""
        samplerate_mhz = 40
        decoder = process.VHSRFDecode(inputfreq=samplerate_mhz, system="NTSC")
        max_hz = 4400000
        max_mhz = max_hz / 1000000
        min_hz = 3400000
        min_mhz = min_hz / 1000000
        wavemax = utils.gen_wave_at_frequency(
            max_mhz, samplerate_mhz, decoder.blocklen // 2
        )
        wavemin = utils.gen_wave_at_frequency(
            min_mhz, samplerate_mhz, decoder.blocklen // 2
        )
        wave = np.concatenate((wavemax, wavemin))
        demod = decoder.demodblock(data=wave)["video"]["demod"]

        max_ire_abs = decoder.iretohz(100)
        sync_ire_abs = decoder.iretohz(decoder.SysParams["vsync_ire"])
        print("max: ", max_ire_abs)
        print("min_ire: ", decoder.SysParams["vsync_ire"])
        print("min: ", sync_ire_abs)

        max_demod = demod[1000 : (decoder.blocklen // 2) - 1000]
        np.testing.assert_allclose(
            max_demod, np.full(len(max_demod), max_hz), rtol=1e-04, atol=20
        )
        min_demod = demod[(decoder.blocklen // 2) + 1000 : decoder.blocklen - 1000]
        # Demodulated signal fluctuates more at the sync end than at the top end,
        # so allowing a little more tolerance here.
        np.testing.assert_allclose(
            min_demod, np.full(len(min_demod), min_hz), rtol=1e-04, atol=50
        )


def test_sync(filename, num_pulses=None, blank_approx=None, sync_approx=None):
    import lddecode.core as ldd
    from vhsdecode.field import FieldPALVHS
    import logging
    import math

    samplerate_mhz = 40

    ldd.logger = logging.getLogger("test")
    ldd.logger.setLevel(5)
    ldd.logger.info("test")

    # process.VHSDecode("infile", "outfile", ,inputfreq=samplerate_mhz, system="PAL", tape_format="VHS")
    rfdecoder = process.VHSRFDecode(
        inputfreq=samplerate_mhz, system="PAL", tape_format="VHS"
    )

    demod_05_data = np.loadtxt(filename)

    data_stub = {}
    data_stub["input"] = np.zeros(5)
    data_stub["video"] = {}
    data_stub["video"]["demod"] = np.zeros_like(demod_05_data)
    data_stub["video"]["demod_05"] = demod_05_data

    field = FieldPALVHS(rfdecoder, data_stub)

    pulses = field.get_pulses()

    if num_pulses:
        assert len(pulses) == num_pulses

    measured_sync = field.sync_tip_level
    measured_blank = field.blanking_level

    if blank_approx:
        assert math.isclose(measured_blank, blank_approx)
    if sync_approx:
        assert math.isclose(measured_sync, sync_approx)

    return True


class SyncTest(unittest.TestCase):
    def test_sync_pal_good(self):
        blank = 4130000
        sync = 3840000
        print("pal good")
        test_sync(
            "PAL_GOOD.txt.gz", num_pulses=458, blank_approx=blank, sync_approx=sync
        )

    def test_sync_pal_noisy(self):
        blank = 4121000
        sync = 3800000
        print("pal noisy")
        test_sync("PAL_NOISY.txt.gz", blank_approx=blank, sync_approx=sync)


class ZCTest(unittest.TestCase):
    def test_calczc(self):
        data = np.array([8.0, 4.0, 1.0, 8.0, 1.0, 4.0, 8.0])
        data2 = np.array([8.0, 4.0, 4.0, 8.0, 30.0, 99.0, 8.0])

        _ = c_calczc(data, 0, 3.0, edge=-1, count=len(data))
        _ = c_calczc(data, 2, 3.0, edge=1, count=len(data))

        _ = c_calczc(data2, 0, 3.0, edge=-1, count=len(data))
        zc4 = c_calczc(data2, 0, 3.0, edge=1, count=len(data))
        assert zc4 is None

        # fig, (ax1, ax2) = plt.subplots(2, 1, sharex=True)

        # ax1.plot(data)
        # ax1.axvline(zc, color="#FF0000")
        # ax1.axvline(zc2, color="#FFFF00")
        # ax1.axhline(3.0)
        # ax2.plot(data2)
        # ax2.axvline(zc3)
        # plt.show()


class RustNumpyMath(unittest.TestCase):
    def test_rust_angle(self):
        from vhsd_rust import complex_angle_py

        loaded = np.load("hilbert_data.npz")
        complex_hilbert_data = loaded["data"]
        a = np.angle(complex_hilbert_data)
        b = complex_angle_py(complex_hilbert_data)
        assert (a == b).all()

    def test_rust_unwrap(self):
        from vhsd_rust import unwrap_angles

        # phase = [0.0, 0.78539816, 1.57079633, 5.49778714, 6.28318531]

        loaded = np.load("hilbert_data.npz")
        # complex_hilbert_data = loaded["data"]
        data = np.angle(loaded["data"])
        a = np.unwrap(data)
        b = unwrap_angles(np.array(data))

        # print(a - b)
        assert np.isclose(b, a, atol=1e-15, rtol=1e-13).all()

    def test_rust_diff(self):
        from vhsdecode.hilbert import diff_forward
        from vhsd_rust import diff_forward_in_place

        loaded = np.load("hilbert_data.npz")
        data = np.angle(loaded["data"])
        output_ediff = np.ediff1d(data, to_begin=0)
        output_cython = diff_forward(data)
        output_rust = np.copy(data)
        diff_forward_in_place(output_rust)
        assert (output_ediff == output_cython).all()
        assert (output_rust == output_cython).all()


class LevelDetect(unittest.TestCase):
    def test_1(self):
        import lddecode.core as ldd
        from vhsdecode.field import FieldPALVHS
        import logging

        ldd.logger = logging.getLogger("test")
        ldd.logger.setLevel(5)
        ldd.logger.info("test")
        demod_05_data = np.loadtxt("PAL_GOOD.txt.gz")

        rfdecoder = process.VHSRFDecode(
            inputfreq=40, system="PAL", tape_format="VHS"
        )

        data_stub = {}
        data_stub["input"] = np.zeros(5)
        data_stub["video"] = {}
        data_stub["video"]["demod"] = np.zeros_like(demod_05_data)
        data_stub["video"]["demod_05"] = demod_05_data

        field = FieldPALVHS(rfdecoder, data_stub)

        _ = field.get_pulses(True)
        sync_level = field.sync_tip_level
        blank_level = field.blanking_level
        print("sync level: ", sync_level)
        print("blank level: ", blank_level)


if __name__ == "__main__":
    unittest.main()