import datetime
import os
import shutil
import tempfile
import unittest

import numpy as np
import pandas as pd

import wfdb


class TestRecord(unittest.TestCase):
    """
    Test read and write of single segment WFDB records, including
    PhysioNet streaming.

    Target files created using the original WFDB Software Package
    version 10.5.24

    """

    # ----------------------- 1. Basic Tests -----------------------#

    def test_1a(self):
        """
        Format 16, entire signal, digital.

        Target file created with:
            rdsamp -r sample-data/test01_00s | cut -f 2- > record-1a
        """
        record = wfdb.rdrecord(
            "sample-data/test01_00s", physical=False, return_res=16
        )
        sig = record.d_signal
        sig_target = np.genfromtxt("tests/target-output/record-1a")

        # Compare data streaming from Physionet
        record_pn = wfdb.rdrecord(
            "test01_00s", pn_dir="macecgdb", physical=False, return_res=16
        )

        # Test file writing
        record_2 = wfdb.rdrecord(
            "sample-data/test01_00s", physical=False, return_res=16
        )
        record_2.sig_name = ["ECG_1", "ECG_2", "ECG_3", "ECG_4"]
        record_2.wrsamp(write_dir=self.temp_path)
        record_write = wfdb.rdrecord(
            os.path.join(self.temp_path, "test01_00s"),
            physical=False,
            return_res=16,
        )

        assert np.array_equal(sig, sig_target)
        assert record.__eq__(record_pn)
        assert record_2.__eq__(record_write)

    def test_1b(self):
        """
        Format 16, byte offset, selected duration, selected channels,
        physical.

        Target file created with:
            rdsamp -r sample-data/a103l -f 50 -t 160 -s 2 0 -P | cut -f 2- > record-1b
        """
        sig, fields = wfdb.rdsamp(
            "sample-data/a103l", sampfrom=12500, sampto=40000, channels=[2, 0]
        )
        sig_round = np.round(sig, decimals=8)
        sig_target = np.genfromtxt("tests/target-output/record-1b")

        # Compare data streaming from Physionet
        sig_pn, fields_pn = wfdb.rdsamp(
            "a103l",
            pn_dir="challenge-2015/training",
            sampfrom=12500,
            sampto=40000,
            channels=[2, 0],
        )

        # Option of selecting channels by name
        sig_named, fields_named = wfdb.rdsamp(
            "sample-data/a103l",
            sampfrom=12500,
            sampto=40000,
            channel_names=["PLETH", "II"],
        )

        assert np.array_equal(sig_round, sig_target)
        assert np.array_equal(sig, sig_pn) and fields == fields_pn
        assert np.array_equal(sig, sig_named) and fields == fields_named

    def test_1c(self):
        """
        Format 16, byte offset, selected duration, selected channels,
        digital, expanded format.

        Target file created with:
            rdsamp -r sample-data/a103l -f 80 -s 0 1 | cut -f 2- > record-1c
        """
        record = wfdb.rdrecord(
            "sample-data/a103l",
            sampfrom=20000,
            channels=[0, 1],
            physical=False,
            smooth_frames=False,
        )
        # convert expanded to uniform array
        sig = np.zeros((record.sig_len, record.n_sig))
        for i in range(record.n_sig):
            sig[:, i] = record.e_d_signal[i]

        sig_target = np.genfromtxt("tests/target-output/record-1c")

        # Compare data streaming from Physionet
        record_pn = wfdb.rdrecord(
            "a103l",
            pn_dir="challenge-2015/training",
            sampfrom=20000,
            channels=[0, 1],
            physical=False,
            smooth_frames=False,
        )

        # Test file writing
        record.wrsamp(write_dir=self.temp_path, expanded=True)
        record_write = wfdb.rdrecord(
            os.path.join(self.temp_path, "a103l"),
            physical=False,
            smooth_frames=False,
        )

        assert np.array_equal(sig, sig_target)
        assert record.__eq__(record_pn)
        assert record.__eq__(record_write)

    def test_1d(self):
        """
        Format 80, selected duration, selected channels, physical

        Target file created with:
            rdsamp -r sample-data/3000003_0003 -f 1 -t 8 -s 1 -P | cut -f 2- > record-1d
        """
        sig, fields = wfdb.rdsamp(
            "sample-data/3000003_0003", sampfrom=125, sampto=1000, channels=[1]
        )
        sig_round = np.round(sig, decimals=8)
        sig_target = np.genfromtxt("tests/target-output/record-1d")
        sig_target = sig_target.reshape(len(sig_target), 1)

        # Compare data streaming from Physionet
        sig_pn, fields_pn = wfdb.rdsamp(
            "3000003_0003",
            pn_dir="mimic3wdb/30/3000003/",
            sampfrom=125,
            sampto=1000,
            channels=[1],
        )

        assert np.array_equal(sig_round, sig_target)
        assert np.array_equal(sig, sig_pn) and fields == fields_pn

    def test_1e(self):
        """
        Format 24, entire signal, digital.

        Target file created with:
            rdsamp -r sample-data/n8_evoked_raw_95_F1_R9 | cut -f 2- |
            gzip -9 -n > record-1e.gz
        """
        record = wfdb.rdrecord(
            "sample-data/n8_evoked_raw_95_F1_R9", physical=False
        )
        sig = record.d_signal
        sig_target = np.genfromtxt("tests/target-output/record-1e.gz")
        sig_target[sig_target == -32768] = -(2**23)

        # Compare data streaming from Physionet
        record_pn = wfdb.rdrecord(
            "n8_evoked_raw_95_F1_R9", physical=False, pn_dir="earndb/raw/N8"
        )

        # Test file writing
        record_2 = wfdb.rdrecord(
            "sample-data/n8_evoked_raw_95_F1_R9", physical=False
        )
        record_2.wrsamp(write_dir=self.temp_path)
        record_write = wfdb.rdrecord(
            os.path.join(self.temp_path, "n8_evoked_raw_95_F1_R9"),
            physical=False,
        )

        assert np.array_equal(sig, sig_target)
        assert record.__eq__(record_pn)
        assert record_2.__eq__(record_write)

    def test_1f(self):
        """
        All binary formats, multiple signal files in one record.

        Target file created with:
            rdsamp -r sample-data/binformats | cut -f 2- |
            gzip -9 -n > record-1f.gz
        """
        record = wfdb.rdrecord("sample-data/binformats", physical=False)
        sig_target = np.genfromtxt("tests/target-output/record-1f.gz")

        for n, name in enumerate(record.sig_name):
            np.testing.assert_array_equal(
                record.d_signal[:, n], sig_target[:, n], "Mismatch in %s" % name
            )

        for sampfrom in range(0, 3):
            for sampto in range(record.sig_len - 3, record.sig_len):
                record_2 = wfdb.rdrecord(
                    "sample-data/binformats",
                    physical=False,
                    sampfrom=sampfrom,
                    sampto=sampto,
                )
                for n, name in enumerate(record.sig_name):
                    if record.fmt[n] != "8":
                        np.testing.assert_array_equal(
                            record_2.d_signal[:, n],
                            sig_target[sampfrom:sampto, n],
                            "Mismatch in %s" % name,
                        )

    def test_read_write_flac(self):
        """
        All FLAC formats, multiple signal files in one record.

        Target file created with:
            rdsamp -r sample-data/flacformats | cut -f 2- |
            gzip -9 -n > record-flac.gz
        """
        record = wfdb.rdrecord("sample-data/flacformats", physical=False)
        sig_target = np.genfromtxt("tests/target-output/record-flac.gz")

        for n, name in enumerate(record.sig_name):
            np.testing.assert_array_equal(
                record.d_signal[:, n], sig_target[:, n], f"Mismatch in {name}"
            )

        for sampfrom in range(0, 3):
            for sampto in range(record.sig_len - 3, record.sig_len):
                record_2 = wfdb.rdrecord(
                    "sample-data/flacformats",
                    physical=False,
                    sampfrom=sampfrom,
                    sampto=sampto,
                )
                for n, name in enumerate(record.sig_name):
                    np.testing.assert_array_equal(
                        record_2.d_signal[:, n],
                        sig_target[sampfrom:sampto, n],
                        f"Mismatch in {name}",
                    )

        # Test file writing
        record.wrsamp(write_dir=self.temp_path)
        record_write = wfdb.rdrecord(
            os.path.join(self.temp_path, "flacformats"),
            physical=False,
        )
        assert record == record_write

    def test_read_write_flac_multifrequency(self):
        """
        Format 516 with multiple signal files and variable samples per frame.
        """
        # Check that we can read a record and write it out again
        record = wfdb.rdrecord(
            "sample-data/mixedsignals",
            physical=False,
            smooth_frames=False,
        )
        record.wrsamp(write_dir=self.temp_path, expanded=True)

        # Check that result matches the original
        record = wfdb.rdrecord("sample-data/mixedsignals", smooth_frames=False)
        record_write = wfdb.rdrecord(
            os.path.join(self.temp_path, "mixedsignals"),
            smooth_frames=False,
        )
        assert record == record_write

    def test_read_flac_longduration(self):
        """
        Three signals multiplexed in a FLAC file, over 2**24 samples.

        Input file created with:
            yes 25 50 75 | head -5600000 |
            wrsamp -O 508 -o flac_3_constant 0 1 2

        Note that the total number of samples (across the three
        channels) exceeds 2**24.  There is a bug in libsndfile that
        causes it to break if we try to read more than 2**24 total
        samples at a time, when the number of channels is not a power
        of two.
        """
        record = wfdb.rdrecord("sample-data/flac_3_constant")
        sig_target = np.repeat(
            np.array([[0.125, 0.25, 0.375]], dtype="float64"),
            5600000,
            axis=0,
        )
        np.testing.assert_array_equal(record.p_signal, sig_target)

    # ------------------ 2. Special format records ------------------ #

    def test_2a(self):
        """
        Format 212, entire signal, physical.

        Target file created with:
            rdsamp -r sample-data/100 -P | cut -f 2- > record-2a
        """
        sig, fields = wfdb.rdsamp("sample-data/100")
        sig_round = np.round(sig, decimals=8)
        sig_target = np.genfromtxt("tests/target-output/record-2a")

        # Compare data streaming from Physionet
        sig_pn, fields_pn = wfdb.rdsamp("100", pn_dir="mitdb")
        # This comment line was manually added and is not present in the
        # original PhysioNet record
        del fields["comments"][0]

        assert np.array_equal(sig_round, sig_target)
        assert np.array_equal(sig, sig_pn) and fields == fields_pn

    def test_2b(self):
        """
        Format 212, selected duration, selected channel, digital.

        Target file created with:
            rdsamp -r sample-data/100 -f 0.002 -t 30 -s 1 | cut -f 2- > record-2b
        """
        record = wfdb.rdrecord(
            "sample-data/100",
            sampfrom=1,
            sampto=10800,
            channels=[1],
            physical=False,
        )
        sig = record.d_signal
        sig_target = np.genfromtxt("tests/target-output/record-2b")
        sig_target = sig_target.reshape(len(sig_target), 1)

        # Compare data streaming from Physionet
        record_pn = wfdb.rdrecord(
            "100",
            sampfrom=1,
            sampto=10800,
            channels=[1],
            physical=False,
            pn_dir="mitdb",
        )
        # This comment line was manually added and is not present in the
        # original PhysioNet record
        del record.comments[0]

        # Option of selecting channels by name
        record_named = wfdb.rdrecord(
            "sample-data/100",
            sampfrom=1,
            sampto=10800,
            channel_names=["V5"],
            physical=False,
        )
        del record_named.comments[0]

        # Test file writing
        record.wrsamp(write_dir=self.temp_path)
        record_write = wfdb.rdrecord(
            os.path.join(self.temp_path, "100"),
            physical=False,
        )

        assert np.array_equal(sig, sig_target)
        assert record.__eq__(record_pn)
        assert record.__eq__(record_named)
        assert record.__eq__(record_write)

    def test_2c(self):
        """
        Format 212, entire signal, physical, odd sampled record.

        Target file created with:
            rdsamp -r sample-data/100_3chan -P | cut -f 2- > record-2c
        """
        record = wfdb.rdrecord("sample-data/100_3chan")
        sig_round = np.round(record.p_signal, decimals=8)
        sig_target = np.genfromtxt("tests/target-output/record-2c")

        # Test file writing
        record.d_signal = record.adc()
        record.wrsamp(write_dir=self.temp_path)
        record_write = wfdb.rdrecord(os.path.join(self.temp_path, "100_3chan"))
        record.d_signal = None

        assert np.array_equal(sig_round, sig_target)
        assert record.__eq__(record_write)

    def test_2d(self):
        """
        Format 310, selected duration, digital
        Target file created with:
            rdsamp -r sample-data/3000003_0003 -f 0 -t 8.21 | cut -f 2- | wrsamp -o 310derive -O 310
            rdsamp -r 310derive -f 0.007 | cut -f 2- > record-2d
        """
        record = wfdb.rdrecord(
            "sample-data/310derive", sampfrom=2, physical=False
        )
        sig = record.d_signal
        sig_target = np.genfromtxt("tests/target-output/record-2d")
        assert np.array_equal(sig, sig_target)

    def test_2e(self):
        """
        Format 311, selected duration, physical.

        Target file created with:
            rdsamp -r sample-data/3000003_0003 -f 0 -t 8.21 -s 1 | cut -f 2- | wrsamp -o 311derive -O 311
            rdsamp -r 311derive -f 0.005 -t 3.91 -P | cut -f 2- > record-2e
        """
        sig, fields = wfdb.rdsamp(
            "sample-data/311derive", sampfrom=1, sampto=978
        )
        sig = np.round(sig, decimals=8)
        sig_target = np.genfromtxt("tests/target-output/record-2e")
        sig_target = sig_target.reshape([977, 1])
        assert np.array_equal(sig, sig_target)

    # --------------------- 3. Multi-dat records --------------------- #

    def test_3a(self):
        """
        Multi-dat, entire signal, digital
        Target file created with:
            rdsamp -r sample-data/s0010_re | cut -f 2- > record-3a
        """
        record = wfdb.rdrecord("sample-data/s0010_re", physical=False)
        sig = record.d_signal
        sig_target = np.genfromtxt("tests/target-output/record-3a")

        # Compare data streaming from Physionet
        record_pn = wfdb.rdrecord(
            "s0010_re", physical=False, pn_dir="ptbdb/patient001"
        )

        # Test file writing
        record.wrsamp(write_dir=self.temp_path)
        record_write = wfdb.rdrecord(
            os.path.join(self.temp_path, "s0010_re"),
            physical=False,
        )

        assert np.array_equal(sig, sig_target)
        assert record.__eq__(record_pn)
        assert record.__eq__(record_write)

    def test_3b(self):
        """
        Multi-dat, selected duration, selected channels, physical.

        Target file created with:
            rdsamp -r sample-data/s0010_re -f 5 -t 38 -P -s 13 0 4 8 3 | cut -f 2- > record-3b
        """
        sig, fields = wfdb.rdsamp(
            "sample-data/s0010_re",
            sampfrom=5000,
            sampto=38000,
            channels=[13, 0, 4, 8, 3],
        )
        sig_round = np.round(sig, decimals=8)
        sig_target = np.genfromtxt("tests/target-output/record-3b")

        # Compare data streaming from Physionet
        sig_pn, fields_pn = wfdb.rdsamp(
            "s0010_re",
            sampfrom=5000,
            pn_dir="ptbdb/patient001",
            sampto=38000,
            channels=[13, 0, 4, 8, 3],
        )

        assert np.array_equal(sig_round, sig_target)
        assert np.array_equal(sig, sig_pn) and fields == fields_pn

    # -------------- 4. Skew and multiple samples/frame -------------- #

    def test_4a(self):
        """
        Format 16, multi-samples per frame, skew, digital.

        Target file created with:
            rdsamp -r sample-data/test01_00s_skewframe | cut -f 2- > record-4a
        """
        record = wfdb.rdrecord(
            "sample-data/test01_00s_skewframe", physical=False
        )
        sig = record.d_signal
        # The WFDB library rdsamp does not return the final N samples for all
        # channels due to the skew. The WFDB python rdsamp does return the final
        # N samples, filling in NANs for end of skewed channels only.
        sig = sig[:-3, :]

        sig_target = np.genfromtxt("tests/target-output/record-4a")

        # Test file writing. Multiple samples per frame and skew.
        # Have to read all the samples in the record, ignoring skew
        record_no_skew = wfdb.rdrecord(
            "sample-data/test01_00s_skewframe",
            physical=False,
            smooth_frames=False,
            ignore_skew=True,
        )
        record_no_skew.wrsamp(write_dir=self.temp_path, expanded=True)
        # Read the written record
        record_write = wfdb.rdrecord(
            os.path.join(self.temp_path, "test01_00s_skewframe"),
            physical=False,
        )

        assert np.array_equal(sig, sig_target)
        assert record.__eq__(record_write)

    def test_4b(self):
        """
        Format 12, multi-samples per frame, skew, entire signal, digital.

        Target file created with:
            rdsamp -r sample-data/03700181 | cut -f 2- > record-4b
        """
        record = wfdb.rdrecord("sample-data/03700181", physical=False)
        sig = record.d_signal
        # The WFDB library rdsamp does not return the final N samples for all
        # channels due to the skew.
        sig = sig[:-4, :]
        # The WFDB python rdsamp does return the final N samples, filling in
        # NANs for end of skewed channels only.
        sig_target = np.genfromtxt("tests/target-output/record-4b")

        # Compare data streaming from Physionet
        record_pn = wfdb.rdrecord(
            "03700181", physical=False, pn_dir="mimicdb/037"
        )

        # Test file writing. Multiple samples per frame and skew.
        # Have to read all the samples in the record, ignoring skew
        record_no_skew = wfdb.rdrecord(
            "sample-data/03700181",
            physical=False,
            smooth_frames=False,
            ignore_skew=True,
        )
        record_no_skew.wrsamp(write_dir=self.temp_path, expanded=True)
        # Read the written record
        record_write = wfdb.rdrecord(
            os.path.join(self.temp_path, "03700181"),
            physical=False,
        )

        assert np.array_equal(sig, sig_target)
        assert record.__eq__(record_pn)
        assert record.__eq__(record_write)

    def test_4c(self):
        """
        Format 12, multi-samples per frame, skew, selected suration,
        selected channels, physical.

        Target file created with:
            rdsamp -r sample-data/03700181 -f 8 -t 128 -s 0 2 -P | cut -f 2- > record-4c
        """
        sig, fields = wfdb.rdsamp(
            "sample-data/03700181", channels=[0, 2], sampfrom=1000, sampto=16000
        )
        sig_round = np.round(sig, decimals=8)
        sig_target = np.genfromtxt("tests/target-output/record-4c")

        # Compare data streaming from Physionet
        sig_pn, fields_pn = wfdb.rdsamp(
            "03700181",
            pn_dir="mimicdb/037",
            channels=[0, 2],
            sampfrom=1000,
            sampto=16000,
        )

        # Test file writing. Multiple samples per frame and skew.
        # Have to read all the samples in the record, ignoring skew
        record_no_skew = wfdb.rdrecord(
            "sample-data/03700181",
            physical=False,
            smooth_frames=False,
            ignore_skew=True,
        )
        record_no_skew.wrsamp(write_dir=self.temp_path, expanded=True)
        # Read the written record
        writesig, writefields = wfdb.rdsamp(
            os.path.join(self.temp_path, "03700181"),
            channels=[0, 2],
            sampfrom=1000,
            sampto=16000,
        )

        assert np.array_equal(sig_round, sig_target)
        assert np.array_equal(sig, sig_pn) and fields == fields_pn
        assert np.array_equal(sig, writesig) and fields == writefields

    def test_4d(self):
        """
        Format 16, multi-samples per frame, skew, read expanded signals

        Target file created with:
            rdsamp -r sample-data/test01_00s_skewframe -P -H | cut -f 2- > record-4d
        """
        record = wfdb.rdrecord(
            "sample-data/test01_00s_skewframe", smooth_frames=False
        )

        # Upsample the channels with lower samples/frame
        expandsig = np.zeros((7994, 3))
        expandsig[:, 0] = np.repeat(record.e_p_signal[0][:-3], 2)
        expandsig[:, 1] = record.e_p_signal[1][:-6]
        expandsig[:, 2] = np.repeat(record.e_p_signal[2][:-3], 2)

        sig_round = np.round(expandsig, decimals=8)
        sig_target = np.genfromtxt("tests/target-output/record-4d")

        assert np.array_equal(sig_round, sig_target)

    def test_write_smoothed(self):
        """
        Test writing a record after reading with smooth_frames
        """
        record = wfdb.rdrecord(
            "sample-data/drive02",
            physical=False,
            smooth_frames=True,
        )
        record.wrsamp(write_dir=self.temp_path)
        record2 = wfdb.rdrecord(
            os.path.join(self.temp_path, "drive02"),
            physical=False,
        )
        np.testing.assert_array_equal(record.d_signal, record2.d_signal)

    def test_to_dataframe(self):
        record = wfdb.rdrecord("sample-data/test01_00s")
        df = record.to_dataframe()

        self.assertEqual(record.sig_name, list(df.columns))
        self.assertEqual(len(df), record.sig_len)
        self.assertEqual(df.index[0], pd.Timedelta(0))
        self.assertEqual(
            df.index[-1],
            pd.Timedelta(seconds=1 / record.fs * (record.sig_len - 1)),
        )
        assert np.array_equal(record.p_signal, df.values)

    def test_header_with_non_utf8(self):
        """
        Ignores non-utf8 characters in the header part.
        """
        record = wfdb.rdrecord("sample-data/test_generator_2")
        sig_units_target = [
            "uV",
            "uV",
            "uV",
            "uV",
            "uV",
            "uV",
            "uV",
            "uV",
            "mV",
            "mV",
            "uV",
            "mV",
        ]
        assert record.units.__eq__(sig_units_target)

    @classmethod
    def setUpClass(cls):
        cls.temp_directory = tempfile.TemporaryDirectory()
        cls.temp_path = cls.temp_directory.name

    @classmethod
    def tearDownClass(cls):
        cls.temp_directory.cleanup()


class TestMultiRecord(unittest.TestCase):
    """
    Test read and write of multi segment WFDB records, including
    PhysioNet streaming.

    Target files created using the original WFDB Software Package
    version 10.5.24

    """

    def test_multi_fixed_a(self):
        """
        Multi-segment, fixed layout, read entire signal.

        Target file created with:
            rdsamp -r sample-data/multi-segment/fixed1/v102s -P | cut -f 2- > record-multi-fixed-a
        """
        record = wfdb.rdrecord("sample-data/multi-segment/fixed1/v102s")
        sig_round = np.round(record.p_signal, decimals=8)
        sig_target = np.genfromtxt("tests/target-output/record-multi-fixed-a")

        np.testing.assert_equal(sig_round, sig_target)

    def test_multi_fixed_b(self):
        """
        Multi-segment, fixed layout, selected duration, samples read
        from one segment.

        Target file created with:
            rdsamp -r sample-data/multi-segment/fixed1/v102s -t s75000 -P | cut -f 2- > record-multi-fixed-b
        """
        record = wfdb.rdrecord(
            "sample-data/multi-segment/fixed1/v102s", sampto=75000
        )
        sig_round = np.round(record.p_signal, decimals=8)
        sig_target = np.genfromtxt("tests/target-output/record-multi-fixed-b")

        np.testing.assert_equal(sig_round, sig_target)

    def test_multi_fixed_c(self):
        """
        Multi-segment, fixed layout, selected duration and channels,
        samples read from multiple segments

        Target file created with:
            rdsamp -r sample-data/multi-segment/fixed1/v102s -f s70000 -t s80000 -s 1 0 3 -P | cut -f 2- > record-multi-fixed-c
        """
        record = wfdb.rdrecord(
            "sample-data/multi-segment/fixed1/v102s",
            sampfrom=70000,
            sampto=80000,
            channels=[1, 0, 3],
        )
        sig_round = np.round(record.p_signal, decimals=8)
        sig_target = np.genfromtxt("tests/target-output/record-multi-fixed-c")

        # Option of selecting channels by name
        record_named = wfdb.rdrecord(
            "sample-data/multi-segment/fixed1/v102s",
            sampfrom=70000,
            sampto=80000,
            channel_names=["V", "II", "RESP"],
        )

        np.testing.assert_equal(sig_round, sig_target)
        assert record.__eq__(record_named)

    def test_multi_fixed_d(self):
        """
        Multi-segment, fixed layout, multi-frequency, selected channels

        Target file created with:
            rdsamp -r sample-data/multi-segment/041s/ -s 3 2 1 -H |
            cut -f 2- | sed s/-32768/-2048/ |
            gzip -9 -n > tests/target-output/record-multi-fixed-d.gz
        """
        record = wfdb.rdrecord(
            "sample-data/multi-segment/041s/041s",
            channels=[3, 2, 1],
            physical=False,
            smooth_frames=False,
        )

        # Convert expanded to uniform array (high-resolution)
        sig = np.zeros((record.sig_len * 4, record.n_sig), dtype=int)
        for i, s in enumerate(record.e_d_signal):
            sig[:, i] = np.repeat(s, len(sig[:, i]) // len(s))

        sig_target = np.genfromtxt(
            "tests/target-output/record-multi-fixed-d.gz"
        )

        record_named = wfdb.rdrecord(
            "sample-data/multi-segment/041s/041s",
            channel_names=["ABP", "V", "I"],
            physical=False,
            smooth_frames=False,
        )

        # Sample values should match the output of rdsamp -H
        np.testing.assert_array_equal(sig, sig_target)
        # channel_names=[...] should give the same result as channels=[...]
        self.assertEqual(record, record_named)

    def test_multi_variable_a(self):
        """
        Multi-segment, variable layout, selected duration, samples read
        from one segment only.

        Target file created with:
            rdsamp -r sample-data/multi-segment/s00001/s00001-2896-10-10-00-31 -f s14428365 -t s14428375 -P | cut -f 2- > record-multi-variable-a
        """
        record = wfdb.rdrecord(
            "sample-data/multi-segment/s00001/s00001-2896-10-10-00-31",
            sampfrom=14428365,
            sampto=14428375,
        )
        sig_round = np.round(record.p_signal, decimals=8)
        sig_target = np.genfromtxt(
            "tests/target-output/record-multi-variable-a"
        )

        np.testing.assert_equal(sig_round, sig_target)

    def test_multi_variable_b(self):
        """
        Multi-segment, variable layout, selected duration, samples read
        from several segments.

        Target file created with:
            rdsamp -r sample-data/multi-segment/s00001/s00001-2896-10-10-00-31 -f s14428364 -t s14428375 -P | cut -f 2- > record-multi-variable-b
        """
        record = wfdb.rdrecord(
            "sample-data/multi-segment/s00001/s00001-2896-10-10-00-31",
            sampfrom=14428364,
            sampto=14428375,
        )
        sig_round = np.round(record.p_signal, decimals=8)
        sig_target = np.genfromtxt(
            "tests/target-output/record-multi-variable-b"
        )

        np.testing.assert_equal(sig_round, sig_target)

    def test_multi_variable_c(self):
        """
        Multi-segment, variable layout, entire signal, physical, expanded

        The reference signal creation cannot be made with rdsamp
        directly because the WFDB c package (10.5.24) applies the single
        adcgain and baseline values from the layout specification
        header, which is undesired in multi-segment signals with
        different adcgain/baseline values across segments.

        Target file created with:
        ```
        for i in {01..18}
        do
            rdsamp -r sample-data/multi-segment/s25047/3234460_00$i -P | cut -f 2- >> record-multi-variable-c
        done
        ```

        Entire signal has 543240 samples.
        - 25740 length empty segment.
        - First 16 segments have same 2 channels, length 420000
        - Last 2 segments have same 3 channels, length 97500

        """
        record = wfdb.rdrecord(
            "sample-data/multi-segment/s25047/s25047-2704-05-04-10-44",
            smooth_frames=False,
        )

        # convert expanded to uniform array and round to 8 digits
        sig_round = np.zeros((record.sig_len, record.n_sig))
        for i in range(record.n_sig):
            sig_round[:, i] = np.round(record.e_p_signal[i], decimals=8)

        sig_target_a = np.full((25740, 3), np.nan)
        sig_target_b = np.concatenate(
            (
                np.genfromtxt(
                    "tests/target-output/record-multi-variable-c",
                    skip_footer=97500,
                ),
                np.full((420000, 1), np.nan),
            ),
            axis=1,
        )
        sig_target_c = np.genfromtxt(
            "tests/target-output/record-multi-variable-c", skip_header=420000
        )
        sig_target = np.concatenate((sig_target_a, sig_target_b, sig_target_c))

        np.testing.assert_equal(sig_round, sig_target)

    def test_multi_variable_d(self):
        """
        Multi-segment, variable layout, selected duration, selected
        channels, digital. There are two channels: PLETH, and II. Their
        fmt, adc_gain, and baseline do not change between the segments.

        Target file created with:
            rdsamp -r sample-data/multi-segment/p000878/p000878-2137-10-26-16-57 -f s3550 -t s7500 -s 0 1 | cut -f 2- | perl -p -e 's/-32768/  -128/g;' > record-multi-variable-d

        """
        record = wfdb.rdrecord(
            "sample-data/multi-segment/p000878/p000878-2137-10-26-16-57",
            sampfrom=3550,
            sampto=7500,
            channels=[0, 1],
            physical=False,
        )
        sig = record.d_signal

        # Compare data streaming from Physionet
        record_pn = wfdb.rdrecord(
            "p000878-2137-10-26-16-57",
            pn_dir="mimic3wdb/matched/p00/p000878/",
            sampfrom=3550,
            sampto=7500,
            channels=[0, 1],
            physical=False,
        )
        sig_target = np.genfromtxt(
            "tests/target-output/record-multi-variable-d"
        )

        # Option of selecting channels by name
        record_named = wfdb.rdrecord(
            "sample-data/multi-segment/p000878/p000878-2137-10-26-16-57",
            sampfrom=3550,
            sampto=7500,
            physical=False,
            channel_names=["PLETH", "II"],
        )

        np.testing.assert_equal(sig, sig_target)
        assert record.__eq__(record_pn)
        assert record.__eq__(record_named)


class TestTimeConversion(unittest.TestCase):
    """
    Test cases for time conversion
    """

    def test_single(self):
        """
        Time conversion for a single-segment record

        This checks the get_frame_number, get_elapsed_time, and
        get_absolute_time methods for a Record object.  The example record
        has no base date defined, so attempting to convert to/from absolute
        time should raise an exception.

        """
        header = wfdb.rdheader("sample-data/test01_00s")

        # these time values should be equivalent
        n = 123 * header.fs
        t = datetime.timedelta(seconds=123)
        self.assertEqual(header.get_frame_number(n), n)
        self.assertEqual(header.get_frame_number(t), n)
        self.assertEqual(header.get_elapsed_time(n), t)
        self.assertEqual(header.get_elapsed_time(t), t)

        # record test01_00s has no base date, so absolute time conversions
        # should fail
        self.assertIsNone(header.base_date)
        d = datetime.datetime(2001, 1, 1, 12, 0, 0)
        self.assertRaises(ValueError, header.get_frame_number, d)
        self.assertRaises(ValueError, header.get_absolute_time, n)
        self.assertRaises(ValueError, header.get_absolute_time, t)

    def test_multisegment_with_date(self):
        """
        Time conversion for a multi-segment record with base date

        This checks the get_frame_number, get_elapsed_time, and
        get_absolute_time methods for a MultiRecord object.  The example
        record has a base date, so we can convert timestamps between all
        three of the supported representations.

        """
        header = wfdb.rdheader(
            "sample-data/multi-segment/p000878/p000878-2137-10-26-16-57"
        )

        # these time values should be equivalent
        n = 123 * header.fs
        t = datetime.timedelta(seconds=123)
        d = t + header.base_datetime
        self.assertEqual(header.get_frame_number(n), n)
        self.assertEqual(header.get_frame_number(t), n)
        self.assertEqual(header.get_frame_number(d), n)
        self.assertEqual(header.get_elapsed_time(n), t)
        self.assertEqual(header.get_elapsed_time(t), t)
        self.assertEqual(header.get_elapsed_time(d), t)
        self.assertEqual(header.get_absolute_time(n), d)
        self.assertEqual(header.get_absolute_time(t), d)
        self.assertEqual(header.get_absolute_time(d), d)


class TestSignal(unittest.TestCase):
    """
    For lower level signal tests

    """

    def test_infer_sig_len(self):
        """
        Infer the signal length of a record without the sig_len header
        Read two headers. The records should be the same.
        """

        record = wfdb.rdrecord("sample-data/drive02")
        record_2 = wfdb.rdrecord("sample-data/drive02-no-len")
        record_2.record_name = record.record_name

        assert record_2.__eq__(record)

        record = wfdb.rdrecord("sample-data/a103l")
        record_2 = wfdb.rdrecord("sample-data/a103l-no-len")
        record_2.record_name = record.record_name

        assert record_2.__eq__(record)

    def test_physical_conversion(self):
        n_sig = 3
        adc_gain = [1.0, 1234.567, 765.4321]
        baseline = [10, 20, -30]
        d_signal = np.repeat(np.arange(-100, 100), 3).reshape(-1, 3)
        e_d_signal = list(d_signal.transpose())
        fmt = ["16", "16", "16"]

        # Test adding or subtracting a small offset (0.01 ADU) to check
        # that we correctly round to the nearest integer
        for offset in (0, -0.01, 0.01):
            p_signal = (d_signal + offset - baseline) / adc_gain
            e_p_signal = list(p_signal.transpose())

            # Test converting p_signal to d_signal

            record = wfdb.Record(
                n_sig=n_sig,
                p_signal=p_signal.copy(),
                adc_gain=adc_gain,
                baseline=baseline,
                fmt=fmt,
            )

            d_signal_converted = record.adc(expanded=False, inplace=False)
            np.testing.assert_array_equal(d_signal_converted, d_signal)

            record.adc(expanded=False, inplace=True)
            np.testing.assert_array_equal(record.d_signal, d_signal)

            # Test converting e_p_signal to e_d_signal

            record = wfdb.Record(
                n_sig=n_sig,
                e_p_signal=[s.copy() for s in e_p_signal],
                adc_gain=adc_gain,
                baseline=baseline,
                fmt=fmt,
            )

            e_d_signal_converted = record.adc(expanded=True, inplace=False)
            self.assertEqual(len(e_d_signal_converted), n_sig)
            for x, y in zip(e_d_signal_converted, e_d_signal):
                np.testing.assert_array_equal(x, y)

            record.adc(expanded=True, inplace=True)
            self.assertEqual(len(record.e_d_signal), n_sig)
            for x, y in zip(record.e_d_signal, e_d_signal):
                np.testing.assert_array_equal(x, y)

            # Test automatic conversion using wfdb.wrsamp()

            wfdb.wrsamp(
                "test_physical_conversion",
                fs=1000,
                sig_name=["X", "Y", "Z"],
                units=["mV", "mV", "mV"],
                p_signal=p_signal,
                adc_gain=adc_gain,
                baseline=baseline,
                fmt=["16", "16", "16"],
                write_dir=self.temp_path,
            )
            record = wfdb.rdrecord(
                os.path.join(self.temp_path, "test_physical_conversion"),
                physical=False,
            )
            np.testing.assert_array_equal(record.d_signal, d_signal)

            record = wfdb.rdrecord(
                os.path.join(self.temp_path, "test_physical_conversion"),
                physical=True,
            )
            for ch, gain in enumerate(adc_gain):
                np.testing.assert_allclose(
                    record.p_signal[:, ch],
                    p_signal[:, ch],
                    rtol=0.0000001,
                    atol=(0.05 / gain),
                )

    @classmethod
    def setUpClass(cls):
        cls.temp_directory = tempfile.TemporaryDirectory()
        cls.temp_path = cls.temp_directory.name

    @classmethod
    def tearDownClass(cls):
        cls.temp_directory.cleanup()


class TestDownload(unittest.TestCase):
    # Test that we can download records with no "dat" file
    # Regression test for https://github.com/MIT-LCP/wfdb-python/issues/118
    def test_dl_database_no_dat_file(self):
        wfdb.dl_database("afdb", self.temp_path, ["00735"])

    # Test that we can download records that *do* have a "dat" file.
    def test_dl_database_with_dat_file(self):
        wfdb.dl_database("afdb", self.temp_path, ["04015"])

    @classmethod
    def setUpClass(cls):
        cls.temp_directory = tempfile.TemporaryDirectory()
        cls.temp_path = cls.temp_directory.name

    @classmethod
    def tearDownClass(cls):
        cls.temp_directory.cleanup()


if __name__ == "__main__":
    unittest.main()
    print("Everything passed!")