_header.py

import datetime
import os
from typing import Any, Dict, List, Optional, Sequence, Tuple

import numpy as np
import pandas as pd

from wfdb.io import _signal, util
from wfdb.io.header import HeaderSyntaxError, rx_record, rx_segment, rx_signal

"""
Notes
-----
In the original WFDB package, certain fields have default values, but
not all of them. Some attributes need to be present for core
functionality, i.e. baseline, whereas others are not essential, yet have
defaults, i.e. base_time.

This inconsistency has likely resulted in the generation of incorrect
files, and general confusion. This library aims to make explicit,
whether certain fields are present in the file, by setting their values
to None if they are not written in, unless the fields are essential, in
which case an actual default value will be set.

The read vs write default values are different for 2 reasons:
1. We want to force the user to be explicit with certain important
   fields when writing WFDB records fields, without affecting
   existing WFDB headers when reading.
2. Certain unimportant fields may be dependencies of other
   important fields. When writing, we want to fill in defaults
   so that the user doesn't need to. But when reading, it should
   be clear that the fields are missing.

"""
int_types = (int, np.int64, np.int32, np.int16, np.int8)
float_types = (float, np.float64, np.float32) + int_types

_SPECIFICATION_COLUMNS = [
    "allowed_types",
    "delimiter",
    "dependency",
    "write_required",
    "read_default",
    "write_default",
]

RECORD_SPECS = pd.DataFrame(
    index=[
        "record_name",
        "n_seg",
        "n_sig",
        "fs",
        "counter_freq",
        "base_counter",
        "sig_len",
        "base_time",
        "base_date",
    ],
    columns=_SPECIFICATION_COLUMNS,
    dtype="object",
    data=[
        [(str,), "", None, True, None, None],  # record_name
        [int_types, "/", "record_name", True, None, None],  # n_seg
        [int_types, " ", "record_name", True, None, None],  # n_sig
        [float_types, " ", "n_sig", True, 250, None],  # fs
        [float_types, "/", "fs", False, None, None],  # counter_freq
        [float_types, "(", "counter_freq", False, None, None],  # base_counter
        [int_types, " ", "fs", True, None, None],  # sig_len
        [
            (datetime.time,),
            " ",
            "sig_len",
            False,
            None,
            "00:00:00",
        ],  # base_time
        [(datetime.date,), " ", "base_time", False, None, None],  # base_date
    ],
)

SIGNAL_SPECS = pd.DataFrame(
    index=[
        "file_name",
        "fmt",
        "samps_per_frame",
        "skew",
        "byte_offset",
        "adc_gain",
        "baseline",
        "units",
        "adc_res",
        "adc_zero",
        "init_value",
        "checksum",
        "block_size",
        "sig_name",
    ],
    columns=_SPECIFICATION_COLUMNS,
    dtype="object",
    data=[
        [(str,), "", None, True, None, None],  # file_name
        [(str,), " ", "file_name", True, None, None],  # fmt
        [int_types, "x", "fmt", False, 1, None],  # samps_per_frame
        [int_types, ":", "fmt", False, None, None],  # skew
        [int_types, "+", "fmt", False, None, None],  # byte_offset
        [float_types, " ", "fmt", True, 200.0, None],  # adc_gain
        [int_types, "(", "adc_gain", True, 0, None],  # baseline
        [(str,), "/", "adc_gain", True, "mV", None],  # units
        [int_types, " ", "adc_gain", False, None, 0],  # adc_res
        [int_types, " ", "adc_res", False, None, 0],  # adc_zero
        [int_types, " ", "adc_zero", False, None, None],  # init_value
        [int_types, " ", "init_value", False, None, None],  # checksum
        [int_types, " ", "checksum", False, None, 0],  # block_size
        [(str,), " ", "block_size", False, None, None],  # sig_name
    ],
)

SEGMENT_SPECS = pd.DataFrame(
    index=["seg_name", "seg_len"],
    columns=_SPECIFICATION_COLUMNS,
    dtype="object",
    data=[
        [(str, list), "", None, True, None, None],  # seg_name
        [(int_types, list), " ", "seg_name", True, None, None],  # seg_len
    ],
)

# Specifications of all WFDB header fields, except for comments
FIELD_SPECS = pd.concat((RECORD_SPECS, SIGNAL_SPECS, SEGMENT_SPECS))


class BaseHeaderMixin(object):
    """
    Mixin class with multi-segment header methods. Inherited by Record and
    MultiRecord classes.

    Attributes
    ----------
    N/A

    """

    def get_write_subset(self, spec_type):
        """
        Get a set of fields used to write the header; either 'record'
        or 'signal' specification fields. Helper function for
        `get_write_fields`. Gets the default required fields, the user
        defined fields, and their dependencies.

        Parameters
        ----------
        spec_type : str
            The set of specification fields desired. Either 'record' or
            'signal'.

        Returns
        -------
        write_fields : list or dict
            For record fields,  returns a list of all fields needed. For
            signal fields, it returns a dictionary of all fields needed,
            with keys = field and value = list of channels that must be
            present for the field.

        """
        if spec_type == "record":
            write_fields = []
            record_specs = RECORD_SPECS.copy()

            # Remove the n_seg requirement for single segment items
            if not hasattr(self, "n_seg"):
                record_specs.drop("n_seg", inplace=True)

            for field in record_specs.index[-1::-1]:
                # Continue if the field has already been included
                if field in write_fields:
                    continue
                # If the field is required by default or has been
                # defined by the user
                if (
                    record_specs.loc[field, "write_required"]
                    or getattr(self, field) is not None
                ):
                    req_field = field
                    # Add the field and its recursive dependencies
                    while req_field is not None:
                        write_fields.append(req_field)
                        req_field = record_specs.loc[req_field, "dependency"]
            # Add comments if any
            if getattr(self, "comments") is not None:
                write_fields.append("comments")

        # signal spec field. Need to return a potentially different list for each channel.
        elif spec_type == "signal":
            # List of lists for each channel
            write_fields = []
            signal_specs = SIGNAL_SPECS.copy()

            for ch in range(self.n_sig):
                # The fields needed for this channel
                write_fields_ch = []
                for field in signal_specs.index[-1::-1]:
                    if field in write_fields_ch:
                        continue

                    item = getattr(self, field)
                    # If the field is required by default or has been defined by the user
                    if signal_specs.loc[field, "write_required"] or (
                        item is not None and item[ch] is not None
                    ):
                        req_field = field
                        # Add the field and its recursive dependencies
                        while req_field is not None:
                            write_fields_ch.append(req_field)
                            req_field = signal_specs.loc[
                                req_field, "dependency"
                            ]

                write_fields.append(write_fields_ch)

            # Convert the list of lists to a single dictionary.
            # keys = field and value = list of channels in which the
            # field is required.
            dict_write_fields = {}

            # For fields present in any channel:
            for field in set(
                [i for write_fields_ch in write_fields for i in write_fields_ch]
            ):
                dict_write_fields[field] = []

                for ch in range(self.n_sig):
                    if field in write_fields[ch]:
                        dict_write_fields[field].append(ch)

            write_fields = dict_write_fields

        return write_fields


class HeaderMixin(BaseHeaderMixin):
    """
    Mixin class with single-segment header methods. Inherited by Record class.

    Attributes
    ----------
    N/A

    """

    def set_defaults(self):
        """
        Set defaults for fields needed to write the header if they have
        defaults.

        Parameters
        ----------
        N/A

        Returns
        -------
        N/A

        Notes
        -----
        - This is NOT called by `rdheader`. It is only automatically
          called by the gateway `wrsamp` for convenience.
        - This is also not called by `wrheader` since it is supposed to
          be an explicit function.
        - This is not responsible for initializing the attributes. That
          is done by the constructor.

        See also `set_p_features` and `set_d_features`.

        """
        rfields, sfields = self.get_write_fields()
        for f in rfields:
            self.set_default(f)
        for f in sfields:
            self.set_default(f)

    def wrheader(self, write_dir="", expanded=True, wfdb_archive=None):
        """
        Write a WFDB header file. The signals are not used. Before
        writing:
        - Get the fields used to write the header for this instance.
        - Check each required field.
        - Check that the fields are cohesive with one another.

        Parameters
        ----------
        write_dir : str, optional
            The output directory in which the header is written.
        expanded : bool, optional
            Whether the header file should include `samps_per_frame` (this
            should only be true if the signal files are written using
            `expanded=True`).

        Returns
        -------
        N/A

        Notes
        -----
        This function does NOT call `set_defaults`. Essential fields
        must be set beforehand.

        """
        # Get all the fields used to write the header
        # sig_write_fields is a dictionary of
        # {field_name:required_channels}
        rec_write_fields, sig_write_fields = self.get_write_fields()
        if not expanded:
            sig_write_fields.pop("samps_per_frame", None)

        # Check the validity of individual fields used to write the header
        # Record specification fields (and comments)
        for field in rec_write_fields:
            self.check_field(field)

        # Signal specification fields.
        for field in sig_write_fields:
            self.check_field(field, required_channels=sig_write_fields[field])

        # Check the cohesion of fields used to write the header
        self.check_field_cohesion(rec_write_fields, list(sig_write_fields))

        # Write the header file using the specified fields
        self.wr_header_file(
            rec_write_fields,
            sig_write_fields,
            write_dir,
            wfdb_archive=wfdb_archive,
        )

    def get_write_fields(self):
        """
        Get the list of fields used to write the header, separating
        record and signal specification fields. Returns the default
        required fields, the user defined fields, and their dependencies.

        Does NOT include `d_signal` or `e_d_signal`.

        Parameters
        ----------
        N/A

        Returns
        -------
        rec_write_fields : list
            Record specification fields to be written. Includes
            'comment' if present.
        sig_write_fields : dict
            Dictionary of signal specification fields to be written,
            with values equal to the channels that need to be present
            for each field.

        """
        # Record specification fields
        rec_write_fields = self.get_write_subset("record")

        # Add comments if any
        if self.comments != None:
            rec_write_fields.append("comments")

        # Get required signal fields if signals are present.
        self.check_field("n_sig")

        if self.n_sig > 0:
            sig_write_fields = self.get_write_subset("signal")
        else:
            sig_write_fields = None

        return rec_write_fields, sig_write_fields

    def _auto_signal_file_names(self):
        fmt = self.fmt or [None] * self.n_sig
        spf = self.samps_per_frame or [None] * self.n_sig
        num_groups = 0
        group_number = []
        prev_fmt = prev_spf = None
        channels_in_group = 0

        for ch_fmt, ch_spf in zip(fmt, spf):
            if ch_fmt != prev_fmt:
                num_groups += 1
                channels_in_group = 0
            elif ch_fmt in ("508", "516", "524"):
                if channels_in_group >= 8 or ch_spf != prev_spf:
                    num_groups += 1
                    channels_in_group = 0
            group_number.append(num_groups)
            channels_in_group += 1
            prev_fmt = ch_fmt
            prev_spf = ch_spf

        if num_groups < 2:
            return [self.record_name + ".dat"] * self.n_sig
        else:
            digits = len(str(group_number[-1]))
            return [
                self.record_name + "_" + str(g).rjust(digits, "0") + ".dat"
                for g in group_number
            ]

    def set_default(self, field):
        """
        Set the object's attribute to its default value if it is missing
        and there is a default. Not responsible for initializing the
        attribute. That is done by the constructor.

        Parameters
        ----------
        field : str
            The desired attribute of the object.

        Returns
        -------
        N/A

        """
        # Record specification fields
        if field in RECORD_SPECS.index:
            # Return if no default to set, or if the field is already
            # present.
            if (
                RECORD_SPECS.loc[field, "write_default"] is None
                or getattr(self, field) is not None
            ):
                return
            setattr(self, field, RECORD_SPECS.loc[field, "write_default"])

        # Signal specification fields
        # Setting entire list default, not filling in blanks in lists.
        elif field in SIGNAL_SPECS.index:
            # Specific dynamic case
            if field == "file_name" and self.file_name is None:
                self.file_name = self._auto_signal_file_names()
                return

            item = getattr(self, field)

            # Return if no default to set, or if the field is already
            # present.
            if (
                SIGNAL_SPECS.loc[field, "write_default"] is None
                or item is not None
            ):
                return

            # Set more specific defaults if possible
            if field == "adc_res" and self.fmt is not None:
                self.adc_res = _signal._fmt_res(self.fmt)
                return

            setattr(
                self,
                field,
                [SIGNAL_SPECS.loc[field, "write_default"]] * self.n_sig,
            )

    def check_field_cohesion(self, rec_write_fields, sig_write_fields):
        """
        Check the cohesion of fields used to write the header.

        Parameters
        ----------
        rec_write_fields : list
            List of record specification fields to write.
        sig_write_fields : dict
            Dictionary of signal specification fields to write, values
            being equal to a list of channels to write for each field.

        Returns
        -------
        N/A

        """
        # If there are no signal specification fields, there is nothing to check.
        if self.n_sig > 0:
            # The length of all signal specification fields must match n_sig
            # even if some of its elements are None.
            for f in sig_write_fields:
                if len(getattr(self, f)) != self.n_sig:
                    raise ValueError(
                        "The length of field: " + f + " must match field n_sig."
                    )

            # Each file_name must correspond to only one fmt, (and only one byte offset if defined).
            datfmts = {}
            for ch in range(self.n_sig):
                if self.file_name[ch] not in datfmts:
                    datfmts[self.file_name[ch]] = self.fmt[ch]
                else:
                    if datfmts[self.file_name[ch]] != self.fmt[ch]:
                        raise ValueError(
                            "Each file_name (dat file) specified must have the same fmt"
                        )

            datoffsets = {}
            if self.byte_offset is not None:
                # At least one byte offset value exists
                for ch in range(self.n_sig):
                    if self.byte_offset[ch] is None:
                        continue
                    if self.file_name[ch] not in datoffsets:
                        datoffsets[self.file_name[ch]] = self.byte_offset[ch]
                    else:
                        if (
                            datoffsets[self.file_name[ch]]
                            != self.byte_offset[ch]
                        ):
                            raise ValueError(
                                "Each file_name (dat file) specified must have the same byte offset"
                            )

    def wr_header_file(
        self, rec_write_fields, sig_write_fields, write_dir, wfdb_archive=None
    ):
        """
        Write a header file using the specified fields. Converts Record
        attributes into appropriate WFDB format strings.

        Parameters
        ----------
        rec_write_fields : list
            List of record specification fields to write.
        sig_write_fields : dict
            Dictionary of signal specification fields to write, values
            being equal to a list of channels to write for each field.
        write_dir : str
            The directory in which to write the header file.
        wfdb_archive : WFDBArchive, optional
            If provided, write the header into this archive instead of to disk.

        Returns
        -------
        N/A

        """
        # Create record specification line
        record_line = ""
        # Traverse the ordered dictionary
        for field in RECORD_SPECS.index:
            # If the field is being used, add it with its delimiter
            if field in rec_write_fields:
                string_field = str(getattr(self, field))

                # Certain fields need extra processing
                if field == "fs" and isinstance(self.fs, float):
                    if round(self.fs, 8) == float(int(self.fs)):
                        string_field = str(int(self.fs))
                elif field == "base_time" and "." in string_field:
                    string_field = string_field.rstrip("0")
                elif field == "base_date":
                    string_field = "/".join(
                        (string_field[8:], string_field[5:7], string_field[:4])
                    )

                record_line += (
                    RECORD_SPECS.loc[field, "delimiter"] + string_field
                )
                # The 'base_counter' field needs to be closed with ')'
                if field == "base_counter":
                    record_line += ")"

        header_lines = [record_line]

        # Create signal specification lines (if any) one channel at a time
        if self.n_sig > 0:
            signal_lines = self.n_sig * [""]
            for ch in range(self.n_sig):
                # Traverse the signal fields
                for field in SIGNAL_SPECS.index:
                    # If the field is being used, add each of its
                    # elements with the delimiter to the appropriate
                    # line
                    if (
                        field in sig_write_fields
                        and ch in sig_write_fields[field]
                    ):
                        signal_lines[ch] += SIGNAL_SPECS.loc[
                            field, "delimiter"
                        ] + str(getattr(self, field)[ch])
                    # The 'baseline' field needs to be closed with ')'
                    if field == "baseline":
                        signal_lines[ch] += ")"

            header_lines += signal_lines

        # Create comment lines (if any)
        if "comments" in rec_write_fields:
            comment_lines = ["# " + comment for comment in self.comments]
            header_lines += comment_lines

        header_str = "\n".join(header_lines) + "\n"
        hea_filename = os.path.basename(self.record_name) + ".hea"

        if wfdb_archive:
            wfdb_archive.write(hea_filename, header_str.encode("utf-8"))
        else:
            util.lines_to_file(hea_filename, write_dir, header_lines)


class MultiHeaderMixin(BaseHeaderMixin):
    """
    Mixin class with multi-segment header methods. Inherited by
    MultiRecord class.

    Attributes
    ----------
    N/A

    """

    n_seg: int
    seg_len: Sequence[int]
    segments: Optional[Sequence]

    def set_defaults(self):
        """
        Set defaults for fields needed to write the header if they have
        defaults. This is NOT called by rdheader. It is only called by the
        gateway wrsamp for convenience. It is also not called by wrheader since
        it is supposed to be an explicit function. Not responsible for
        initializing the attributes. That is done by the constructor.

        Parameters
        ----------
        N/A

        Returns
        -------
        N/A

        """
        for field in self.get_write_fields():
            self.set_default(field)

    def wrheader(self, write_dir="", wfdb_archive=None):
        """
        Write a multi-segment WFDB header file. The signals or segments are
        not used. Before writing:
        - Get the fields used to write the header for this instance.
        - Check each required field.
        - Check that the fields are cohesive with one another.

        Parameters
        ----------
        write_dir : str, optional
            The output directory in which the header is written.

        Returns
        -------
        N/A

        Notes
        -----
        This function does NOT call `set_defaults`. Essential fields
        must be set beforehand.

        """
        # Get all the fields used to write the header
        write_fields = self.get_write_fields()

        # Check the validity of individual fields used to write the header
        for field in write_fields:
            self.check_field(field)

        # Check the cohesion of fields used to write the header
        self.check_field_cohesion()

        # Write the header file using the specified fields
        self.wr_header_file(write_fields, write_dir, wfdb_archive=wfdb_archive)

    def get_write_fields(self):
        """
        Get the list of fields used to write the multi-segment header.

        Parameters
        ----------
        N/A

        Returns
        -------
        write_fields : list
            All the default required fields, the user defined fields,
            and their dependencies.

        """
        # Record specification fields
        write_fields = self.get_write_subset("record")

        # Segment specification fields are all mandatory
        write_fields = write_fields + ["seg_name", "seg_len"]

        # Comments
        if self.comments != None:
            write_fields.append("comments")
        return write_fields

    def set_default(self, field):
        """
        Set a field to its default value if there is a default.

        Parameters
        ----------
        field : str
            The desired attribute of the object.

        Returns
        -------
        N/A

        """
        # Record specification fields
        if field in RECORD_SPECS:
            # Return if no default to set, or if the field is already present.
            if (
                RECORD_SPECS[field].write_def is None
                or getattr(self, field) is not None
            ):
                return
            setattr(self, field, RECORD_SPECS[field].write_def)

    def check_field_cohesion(self):
        """
        Check the cohesion of fields used to write the header.

        Parameters
        ----------
        N/A

        Returns
        -------
        N/A

        """
        # The length of seg_name and seg_len must match n_seg
        for f in ["seg_name", "seg_len"]:
            if len(getattr(self, f)) != self.n_seg:
                raise ValueError(
                    "The length of field: " + f + " does not match field n_seg."
                )

        # Check the sum of the 'seg_len' fields against 'sig_len'
        if np.sum(self.seg_len) != self.sig_len:
            raise ValueError(
                "The sum of the 'seg_len' fields do not match the 'sig_len' field"
            )

    def wr_header_file(self, write_fields, write_dir, wfdb_archive=None):
        """
        Write a header file using the specified fields.

        Parameters
        ----------
        write_fields : list
            All the default required fields, the user defined fields,
            and their dependencies.
        write_dir : str
            The output directory in which the header is written.
        wfdb_archive : WFDBArchive, optional
            If provided, write the header into this archive instead of to disk.

        Returns
        -------
        N/A

        """
        # Create record specification line
        record_line = ""
        # Traverse the ordered dictionary
        for field in RECORD_SPECS.index:
            # If the field is being used, add it with its delimiter
            if field in write_fields:
                record_line += RECORD_SPECS.loc[field, "delimiter"] + str(
                    getattr(self, field)
                )

        header_lines = [record_line]

        # Create segment specification lines
        segment_lines = self.n_seg * [""]
        # For both fields, add each of its elements with the delimiter
        # to the appropriate line
        for field in SEGMENT_SPECS.index:
            for seg_num in range(self.n_seg):
                segment_lines[seg_num] += SEGMENT_SPECS.loc[
                    field, "delimiter"
                ] + str(getattr(self, field)[seg_num])

        header_lines = header_lines + segment_lines

        # Create comment lines (if any)
        if "comments" in write_fields:
            comment_lines = ["# " + comment for comment in self.comments]
            header_lines += comment_lines

        header_str = "\n".join(header_lines) + "\n"
        hea_filename = os.path.basename(self.record_name) + ".hea"

        if wfdb_archive:
            wfdb_archive.write(hea_filename, header_str.encode("utf-8"))
        else:
            util.lines_to_file(hea_filename, write_dir, header_lines)

    def get_sig_segments(self, sig_name=None):
        """
        Get a list of the segment numbers that contain a particular signal
        (or a dictionary of segment numbers for a list of signals).
        Only works if information about the segments has been read in.

        Parameters
        ----------
        sig_name : str, list
            The name of the signals to be segmented.

        Returns
        -------
        sig_dict : dict
            Segments for each desired signal.
        sig_segs : list
            Segments for the desired signal.

        """
        if self.segments is None:
            raise Exception(
                "The MultiRecord's segments must be read in before this method is called. ie. Call rdheader() with rsegment_fieldsments=True"
            )

        # Default value = all signal names.
        if sig_name is None:
            sig_name = self.get_sig_name()

        if isinstance(sig_name, list):
            sig_dict = {}
            for sig in sig_name:
                sig_dict[sig] = self.get_sig_segments(sig)
            return sig_dict
        elif isinstance(sig_name, str):
            sig_segs = []
            for i in range(self.n_seg):
                if (
                    self.seg_name[i] != "~"
                    and sig_name in self.segments[i].sig_name
                ):
                    sig_segs.append(i)
            return sig_segs
        else:
            raise TypeError("sig_name must be a string or a list of strings")

    def get_sig_name(self):
        """
        Get the signal names for the entire record.

        Parameters
        ----------
        N/A

        Returns
        -------
        sig_name : str, list
            The name of the signals to be segmented.

        """
        if self.segments is None:
            raise Exception(
                "The MultiRecord's segments must be read in before this method is called. ie. Call rdheader() with rd_segments=True"
            )

        if self.layout == "fixed":
            for i in range(self.n_seg):
                if self.seg_name[i] != "~":
                    sig_name = self.segments[i].sig_name
                    break
        else:
            sig_name = self.segments[0].sig_name

        return sig_name

    def contained_ranges(self, sig_name: str) -> List[Tuple[int, int]]:
        """
        Given a signal name, return the sample ranges that contain signal values,
        relative to the start of the full record. Does not account for NaNs/missing
        values.

        This function is mainly useful for variable layout records, but can also be
        used for fixed-layout records. Only works if the headers from the individual
        segment records have already been read in.

        Parameters
        ----------
        sig_name : str
            The name of the signal to query.

        Returns
        -------
        ranges : List[Tuple[int, int]]
            Tuple pairs which specify thee sample ranges in which the signal is contained.
            The second value of each tuple pair will be one beyond the signal index.
            eg. A length 1000 signal would generate a tuple of: (0, 1000), allowing
            selection using signal[0:1000].

        """
        if self.segments is None:
            raise Exception(
                "The MultiRecord's segments must be read in before this method is called. ie. Call rdheader() with rd_segments=True"
            )
        ranges = []
        seg_start = 0

        range_start = None

        # TODO: Add shortcut for fixed-layout records

        # Cannot process segments only because missing segments are None
        # and do not contain length information.
        for seg_num in range(self.n_seg):
            seg_len = self.seg_len[seg_num]
            segment = self.segments[seg_num]

            if seg_len == 0:
                continue

            # Open signal range
            if (
                range_start is None
                and segment is not None
                and sig_name in segment.sig_name
            ):
                range_start = seg_start
            # Close signal range
            elif range_start is not None and (
                segment is None or sig_name not in segment.sig_name
            ):
                ranges.append((range_start, seg_start))
                range_start = None

            seg_start += seg_len

        # Account for final segment
        if range_start is not None:
            ranges.append((range_start, seg_start))

        return ranges

    def contained_combined_ranges(
        self,
        sig_names: Sequence[str],
    ) -> List[Tuple[int, int]]:
        """
        Given a collection of signal name, return the sample ranges that
        contain all of the specified signals, relative to the start of the
        full record. Does not account for NaNs/missing values.

        This function is mainly useful for variable layout records, but can also be
        used for fixed-layout records. Only works if the headers from the individual
        segment records have already been read in.

        Parameters
        ----------
        sig_names : List[str]
            The names of the signals to query.

        Returns
        -------
        ranges : List[Tuple[int, int]]
            Tuple pairs which specify thee sample ranges in which the signal is contained.
            The second value of each tuple pair will be one beyond the signal index.
            eg. A length 1000 signal would generate a tuple of: (0, 1000), allowing
            selection using signal[0:1000].

        """
        # TODO: Add shortcut for fixed-layout records

        if len(sig_names) == 0:
            return []

        combined_ranges = self.contained_ranges(sig_names[0])

        if len(sig_names) > 1:
            for name in sig_names[1:]:
                combined_ranges = util.overlapping_ranges(
                    combined_ranges, self.contained_ranges(name)
                )

        return combined_ranges


def wfdb_strptime(time_string: str) -> datetime.time:
    """
    Given a time string in an acceptable WFDB format, return
    a datetime.time object.

    Valid formats: SS, MM:SS, HH:MM:SS, all with and without microsec.

    Parameters
    ----------
    time_string : str
        The time to be converted to a datetime.time object.

    Returns
    -------
    datetime.time object
        The time converted from str format.

    """
    n_colons = time_string.count(":")

    if n_colons == 0:
        time_fmt = "%S"
    elif n_colons == 1:
        time_fmt = "%M:%S"
    elif n_colons == 2:
        time_fmt = "%H:%M:%S"

    if "." in time_string:
        time_fmt += ".%f"

    return datetime.datetime.strptime(time_string, time_fmt).time()


def _parse_record_line(record_line: str) -> dict:
    """
    Extract fields from a record line string into a dictionary.

    Parameters
    ----------
    record_line : str
        The record line contained in the header file

    Returns
    -------
    record_fields : dict
        The fields for the given record line.

    """
    # Dictionary for record fields
    record_fields: Dict[str, Any] = {}

    # Read string fields from record line
    match = rx_record.match(record_line)
    if match is None:
        raise HeaderSyntaxError("invalid syntax in record line")
    (
        record_fields["record_name"],
        record_fields["n_seg"],
        record_fields["n_sig"],
        record_fields["fs"],
        record_fields["counter_freq"],
        record_fields["base_counter"],
        record_fields["sig_len"],
        record_fields["base_time"],
        record_fields["base_date"],
    ) = match.groups()

    for field in RECORD_SPECS.index:
        # Replace empty strings with their read defaults (which are
        # mostly None)
        if record_fields[field] == "":
            record_fields[field] = RECORD_SPECS.loc[field, "read_default"]
        # Typecast non-empty strings for non-string (numerical/datetime)
        # fields
        else:
            if RECORD_SPECS.loc[field, "allowed_types"] == int_types:
                record_fields[field] = int(record_fields[field])
            elif RECORD_SPECS.loc[field, "allowed_types"] == float_types:
                record_fields[field] = float(record_fields[field])
                # cast fs to an int if it is close
                if field == "fs":
                    fs = float(record_fields["fs"])
                    if round(fs, 8) == float(int(fs)):
                        fs = int(fs)
                    record_fields["fs"] = fs
            elif field == "base_time":
                record_fields["base_time"] = wfdb_strptime(
                    record_fields["base_time"]
                )
            elif field == "base_date":
                record_fields["base_date"] = datetime.datetime.strptime(
                    record_fields["base_date"], "%d/%m/%Y"
                ).date()

    # This is not a standard WFDB field, but is useful to set.
    if record_fields["base_date"] and record_fields["base_time"]:
        record_fields["base_datetime"] = datetime.datetime.combine(
            record_fields["base_date"], record_fields["base_time"]
        )

    return record_fields


def _parse_signal_lines(signal_lines):
    """
    Extract fields from a list of signal line strings into a dictionary.

    Parameters
    ----------
    signal_lines : list
        The name of the signal line that will be used to extact fields.

    Returns
    -------
    signal_fields : dict
        The fields for the given signal line.

    """
    n_sig = len(signal_lines)
    # Dictionary for signal fields
    signal_fields = {}

    # Each dictionary field is a list
    for field in SIGNAL_SPECS.index:
        signal_fields[field] = n_sig * [None]

    # Read string fields from signal line
    for ch in range(n_sig):
        match = rx_signal.match(signal_lines[ch])
        if match is None:
            raise HeaderSyntaxError("invalid syntax in signal line")
        (
            signal_fields["file_name"][ch],
            signal_fields["fmt"][ch],
            signal_fields["samps_per_frame"][ch],
            signal_fields["skew"][ch],
            signal_fields["byte_offset"][ch],
            signal_fields["adc_gain"][ch],
            signal_fields["baseline"][ch],
            signal_fields["units"][ch],
            signal_fields["adc_res"][ch],
            signal_fields["adc_zero"][ch],
            signal_fields["init_value"][ch],
            signal_fields["checksum"][ch],
            signal_fields["block_size"][ch],
            signal_fields["sig_name"][ch],
        ) = match.groups()

        for field in SIGNAL_SPECS.index:
            # Replace empty strings with their read defaults (which are mostly None)
            # Note: Never set a field to None. [None]* n_sig is accurate, indicating
            # that different channels can be present or missing.
            if signal_fields[field][ch] == "":
                signal_fields[field][ch] = SIGNAL_SPECS.loc[
                    field, "read_default"
                ]

                # Special case: missing baseline defaults to ADCzero if present
                if field == "baseline" and signal_fields["adc_zero"][ch] != "":
                    signal_fields["baseline"][ch] = int(
                        signal_fields["adc_zero"][ch]
                    )
            # Typecast non-empty strings for numerical fields
            else:
                if SIGNAL_SPECS.loc[field, "allowed_types"] is int_types:
                    signal_fields[field][ch] = int(signal_fields[field][ch])
                elif SIGNAL_SPECS.loc[field, "allowed_types"] is float_types:
                    signal_fields[field][ch] = float(signal_fields[field][ch])
                    # Special case: adc_gain of 0 means 200
                    if (
                        field == "adc_gain"
                        and signal_fields["adc_gain"][ch] == 0
                    ):
                        signal_fields["adc_gain"][ch] = 200.0

    return signal_fields


def _read_segment_lines(segment_lines):
    """
    Extract fields from segment line strings into a dictionary.

    Parameters
    ----------
    segment_line : list
        The name of the segment line that will be used to extact fields.

    Returns
    -------
    segment_fields : dict
        The fields for the given segment line.

    """
    # Dictionary for segment fields
    segment_fields = {}

    # Each dictionary field is a list
    for field in SEGMENT_SPECS.index:
        segment_fields[field] = [None] * len(segment_lines)

    # Read string fields from signal line
    for i in range(len(segment_lines)):
        match = rx_segment.match(segment_lines[i])
        if match is None:
            raise HeaderSyntaxError("invalid syntax in segment line")
        (
            segment_fields["seg_name"][i],
            segment_fields["seg_len"][i],
        ) = match.groups()

        # Typecast strings for numerical field
        if field == "seg_len":
            segment_fields["seg_len"][i] = int(segment_fields["seg_len"][i])

    return segment_fields