python_embed/Lib/site-packages/edfio/edf.py

from __future__ import annotations

import contextlib
import copy
import datetime
import io
import math
import sys
import tempfile
import warnings
from collections.abc import Iterable, Sequence
from decimal import Decimal
from fractions import Fraction
from math import ceil, floor
from pathlib import Path
from typing import ClassVar, Generic, Literal, TypeVar, Union

import numpy as np

from edfio._header_field import (
    decode_date,
    decode_float,
    decode_str,
    decode_time,
    encode_date,
    encode_float,
    encode_int,
    encode_str,
    encode_time,
)
from edfio._lazy_loading import LazyLoader
from edfio.edf_annotations import (
    EdfAnnotation,
    _create_annotations_signal,
    _EdfAnnotationsDataRecord,
)
from edfio.edf_header import (
    AnonymizedDateError,
    Patient,
    Recording,
    _encode_edfplus_date,
)
from edfio.edf_signal import BdfSignal, EdfSignal

if sys.version_info < (3, 11):  # pragma: no cover
    from typing_extensions import Self
else:  # pragma: no cover
    from typing import Self


_Signal = TypeVar("_Signal", bound=Union[EdfSignal, BdfSignal])


class _Base(Generic[_Signal]):
    _header_fields = (
        ("version", 8),
        ("local_patient_identification", 80),
        ("local_recording_identification", 80),
        ("startdate", 8),
        ("starttime", 8),
        ("bytes_in_header_record", 8),
        ("reserved", 44),
        ("num_data_records", 8),
        ("data_record_duration", 8),
        ("num_signals", 4),
    )
    _signals: tuple[_Signal, ...]
    _signal_class: type[_Signal]
    _fmt: ClassVar[Literal["EDF", "BDF"]]

    def __init__(
        self,
        signals: Sequence[_Signal],
        *,
        patient: Patient | None = None,
        recording: Recording | None = None,
        starttime: datetime.time | None = None,
        data_record_duration: float | None = None,
        annotations: Iterable[EdfAnnotation] | None = None,
    ):
        if not signals and not annotations:
            raise ValueError("Edf must contain either signals or annotations")
        if patient is None:
            patient = Patient()
        if recording is None:
            recording = Recording()
        if starttime is None:
            starttime = datetime.time(0, 0, 0)
        if data_record_duration is None:
            data_record_duration = _calculate_data_record_duration(signals)
        elif len(signals) == 0 and data_record_duration != 0:
            raise ValueError(
                "Data record duration must be zero for annotation-only files"
            )

        self._set_data_record_duration(data_record_duration)
        self._set_num_data_records_with_signals(signals)
        self._set_version()
        self.local_patient_identification = patient._to_str()
        self.local_recording_identification = recording._to_str()
        self._set_startdate_with_recording(recording)
        self._starttime = encode_time(starttime.replace(microsecond=0))
        self._set_reserved("")
        if starttime.microsecond and annotations is None:
            warnings.warn(f"Creating {self._fmt}+C to store microsecond starttime.")
        if annotations is not None or starttime.microsecond:
            signals = (
                *signals,
                _create_annotations_signal(
                    annotations if annotations is not None else (),
                    num_data_records=self.num_data_records,
                    data_record_duration=self.data_record_duration,
                    signal_class=self._signal_class,
                    with_timestamps=True,
                    subsecond_offset=starttime.microsecond / 1_000_000,
                ),
            )
            self._set_reserved(f"{self._fmt}+C")
        self._set_signals(signals)
        self._header_encoding = "ascii"

    def __repr__(self) -> str:
        signals_text = f"{len(self.signals)} signal"
        if len(self.signals) != 1:
            signals_text += "s"
        annotations_text = f"{len(self.annotations)} annotation"
        if len(self.annotations) != 1:
            annotations_text += "s"
        return f"<{self.__class__.__name__} {signals_text} {annotations_text}>"

    def _set_version(self) -> None:
        raise NotImplementedError  # pragma: no cover

    def _set_bytes_in_header_record(self, bytes_in_header_record: int) -> None:
        self._bytes_in_header_record = encode_int(bytes_in_header_record, 8)

    def _set_reserved(self, reserved: str) -> None:
        self._reserved = encode_str(reserved, 44)

    def _set_num_data_records(self, num_data_records: int) -> None:
        self._num_data_records = encode_int(num_data_records, 8)

    def _set_data_record_duration(self, data_record_duration: float) -> None:
        self._data_record_duration = encode_float(data_record_duration)

    def _set_num_signals(self, num_signals: int) -> None:
        self._num_signals = encode_int(num_signals, 4)

    @property
    def local_patient_identification(self) -> str:
        """
        Unparsed string representation of the legacy local patient identification.

        See Also
        --------
        patient: Parsed representation, as a :class:`Patient` object.
        """
        return decode_str(self._local_patient_identification, self._header_encoding)

    @local_patient_identification.setter
    def local_patient_identification(self, value: str) -> None:
        self._local_patient_identification = encode_str(value, 80)

    @property
    def local_recording_identification(self) -> str:
        """
        Unparsed string representation of the legacy local recording identification.

        See Also
        --------
        recording: Parsed representation, as a :class:`Recording` object.
        """
        return decode_str(self._local_recording_identification, self._header_encoding)

    @local_recording_identification.setter
    def local_recording_identification(self, value: str) -> None:
        self._local_recording_identification = encode_str(value, 80)

    def _load_data(  # noqa: PLR0912
        self,
        file: Path
        | io.BufferedReader
        | io.BytesIO
        | tempfile.SpooledTemporaryFile[bytes],
        *,
        lazy_load_data: bool,
    ) -> None:
        if lazy_load_data and (
            self._signal_class == BdfSignal or not isinstance(file, Path)
        ):
            raise ValueError(
                "Lazy loading is only supported for local file paths in EDF format, "
                f"got {type(file)} for {self._fmt} format"
            )
        lens = [signal.samples_per_data_record for signal in self._signals]
        datarecord_len = sum(lens)
        truncated = False
        # TODO: Someday maybe we can memmap BDF files, but for now let's just read all
        # https://stackoverflow.com/questions/12080279/how-do-i-create-a-numpy-dtype-that-includes-24-bit-integers
        if self._signal_class is BdfSignal:
            if isinstance(file, Path):
                remaining_bytes = file.stat().st_size - self.bytes_in_header_record
                data_bytes = file.read_bytes()[self.bytes_in_header_record :]
            else:
                data_bytes = file.read()
                remaining_bytes = len(data_bytes)
            actual_records = remaining_bytes // (datarecord_len * 3)
            if actual_records * datarecord_len * 3 < remaining_bytes:
                truncated = True
            datarecords = np.frombuffer(
                data_bytes, dtype=np.uint8, count=3 * actual_records * datarecord_len
            )
            datarecords = datarecords.reshape(-1, 3).astype(np.int32)
            datarecords = (
                datarecords[:, 0] + (datarecords[:, 1] << 8) + (datarecords[:, 2] << 16)
            )
            # 24th bit determines the sign
            datarecords[datarecords >= (1 << 23)] -= 1 << 24
            datarecords.shape = (actual_records, datarecord_len)
        elif not isinstance(file, Path):
            data_bytes = file.read()
            actual_records = len(data_bytes) // (datarecord_len * 2)
            if actual_records * datarecord_len * 2 < len(data_bytes):
                truncated = True
            datarecords = np.frombuffer(
                data_bytes, dtype=np.int16, count=actual_records * datarecord_len
            )
            datarecords.shape = (actual_records, datarecord_len)
        else:
            remaining_bytes = file.stat().st_size - self.bytes_in_header_record
            actual_records = remaining_bytes // (datarecord_len * 2)
            if actual_records * datarecord_len * 2 < remaining_bytes:
                truncated = True
            datarecords = np.memmap(
                file,
                dtype=np.int16,
                mode="r",
                offset=self.bytes_in_header_record,
                shape=(actual_records, datarecord_len),
            )
        if truncated:
            warnings.warn(
                f"Incomplete data record at the end of the {self._fmt} file. "
                "Data was truncated."
            )
        if self.num_data_records != actual_records:
            warnings.warn(
                f"{self._fmt} header indicates {self.num_data_records} data "
                f"records, but file contains {actual_records} records. Updating header."
            )
            self._set_num_data_records(actual_records)
        ends = np.cumsum(lens)
        starts = ends - lens

        for signal, start, end in zip(self._signals, starts, ends):
            if lazy_load_data:
                signal._lazy_loader = LazyLoader(datarecords, start, end)  # type: ignore[type-var, assignment]
            elif signal.label == "BDF Annotations":
                digital = datarecords[:, start:end].flatten()
                digital = digital.view(np.uint8).reshape(-1, 4)[:, :3].flatten()
                signal._digital = digital  # type: ignore[assignment]
            else:
                signal._digital = datarecords[:, start:end].flatten()  # type: ignore[assignment]

    def _read_header(
        self,
        buffer: io.BufferedReader | io.BytesIO | tempfile.SpooledTemporaryFile[bytes],
        header_encoding: str,
    ) -> None:
        self._header_encoding = header_encoding
        for header_name, length in Edf._header_fields:
            setattr(self, "_" + header_name, buffer.read(length))
        self._signals = self._parse_signal_headers(
            buffer.read(256 * self._total_num_signals), header_encoding
        )

    @property
    def signals(self) -> tuple[_Signal, ...]:
        """
        Ordinary signals contained in the recording.

        Annotation signals are excluded. Individual signals can not be removed, added,
        or replaced by modifying this property. Use :meth:`Edf.append_signals`,
        :meth:`Edf.drop_signals`, or :attr:`EdfSignal.data`, respectively.
        """
        return tuple(s for s in self._signals if not s._is_annotation_signal)

    def _set_signals(self, signals: Sequence[_Signal]) -> None:
        signals = tuple(signals)
        for si, signal in enumerate(signals):
            if type(signal) is not self._signal_class:
                raise ValueError(
                    f"_Signal {si} ({signal}) has format {signal._fmt}, but recording is {self._fmt}"
                )
        self._set_num_data_records_with_signals(signals)
        self._signals = signals
        self._set_bytes_in_header_record(256 * (len(signals) + 1))
        self._set_num_signals(len(signals))
        if all(s._is_annotation_signal for s in signals):
            self._set_data_record_duration(0)

    def _set_num_data_records_with_signals(
        self,
        signals: Sequence[_Signal],
    ) -> None:
        if not signals:
            num_data_records = 1
        else:
            signal_durations = [
                round(s._num_samples / s.sampling_frequency, 12) for s in signals
            ]
            if any(v != signal_durations[0] for v in signal_durations[1:]):
                raise ValueError(
                    f"Inconsistent signal durations (in seconds): {signal_durations}"
                )
            num_data_records = _calculate_num_data_records(
                signal_durations[0],
                self.data_record_duration,
            )
            signal_lengths = [s._num_samples for s in signals]
            if any(l % num_data_records for l in signal_lengths):
                raise ValueError(
                    f"Not all signal lengths can be split into {num_data_records} data records: {signal_lengths}"
                )
        self._set_num_data_records(num_data_records)

    def _parse_signal_headers(
        self,
        raw_signal_headers: bytes,
        header_encoding: str,
    ) -> tuple[_Signal, ...]:
        raw_headers_split: dict[str, list[bytes]] = {}
        start = 0
        for header_name, length in EdfSignal._header_fields:
            end = start + length * self._total_num_signals
            raw_header = raw_signal_headers[start:end]
            raw_headers_split[header_name] = [
                raw_header[i : length + i] for i in range(0, len(raw_header), length)
            ]
            start = end
        signals: list[_Signal] = []
        for i in range(self._total_num_signals):
            raw_signal_header = {
                key: raw_headers_split[key][i] for key in raw_headers_split
            }
            try:
                sampling_frequency = (
                    int(raw_signal_header["samples_per_data_record"])
                    / self.data_record_duration
                )
            except ZeroDivisionError:
                if (
                    raw_signal_header["label"].rstrip()
                    == f"{self._fmt} Annotations".encode()
                ):
                    sampling_frequency = 0
            signals.append(
                self._signal_class._from_raw_header(  # type: ignore[arg-type]
                    sampling_frequency,
                    **raw_signal_header,
                    header_encoding=header_encoding,
                )
            )
        return tuple(signals)

    def write(self, target: Path | str | io.BufferedWriter | io.BytesIO) -> None:  # noqa: PLR0912
        """
        Write an Edf to a file or file-like object.

        Parameters
        ----------
        target : Path | str | io.BufferedWriter | io.BytesIO
            The file location (path object or string) or file-like object to write to.
        """
        if self.num_data_records == -1:
            warnings.warn("num_data_records=-1, determining correct value from data")
            num_data_records = _calculate_num_data_records(
                self._signals[0]._num_samples * self._signals[0].sampling_frequency,
                self.data_record_duration,
            )
        else:
            num_data_records = self.num_data_records
        for signal in self._signals:
            signal._set_samples_per_data_record(signal._num_samples // num_data_records)
        header_records = []
        for header_name, _ in Edf._header_fields:
            header_records.append(getattr(self, "_" + header_name))
        for header_name, _ in EdfSignal._header_fields:
            for signal in self._signals:
                header_records.append(getattr(signal, "_" + header_name))
        header_record = b"".join(header_records)

        lens = [s.samples_per_data_record * s._bytes_per_sample for s in self._signals]
        ends = np.cumsum(lens)
        starts = ends - lens
        data_record = np.empty((num_data_records, sum(lens)), dtype=np.uint8)
        for signal, start, end in zip(self._signals, starts, ends):
            if self._signal_class is EdfSignal or signal.label == "BDF Annotations":
                data = signal.digital.view(np.uint8).reshape(num_data_records, -1)
            else:
                data = (
                    signal.digital.view(np.uint8)
                    .reshape(-1, 4)[:, :3]
                    .reshape(num_data_records, -1)
                )
            data_record[:, start:end] = data
        if isinstance(target, str):
            target = Path(target)
        if isinstance(target, io.BufferedWriter):
            target.write(header_record)
            data_record.tofile(target)
        elif isinstance(target, io.BytesIO):
            target.write(header_record)
            target.write(data_record.tobytes())
        else:
            with target.expanduser().open("wb") as file:
                file.write(header_record)
                data_record.tofile(file)

    @property
    def labels(self) -> tuple[str, ...]:
        """
        The labels of all signals contained in the Edf.

        Returns
        -------
        tuple[str, ...]
            The labels, in order of the signals.
        """
        return tuple(s.label for s in self.signals)

    def get_signal(self, label: str) -> _Signal:
        """
        Retrieve a single signal by its label.

        The label has to be unique - a ValueError is raised if it is ambiguous or does
        not exist.

        Parameters
        ----------
        label : str
            A label identifying a single signal

        Returns
        -------
        EdfSignal
            The signal corresponding to the given label.
        """
        count = self.labels.count(label)
        if count == 0:
            raise ValueError(
                f"No signal with label {label!r}, possible options: {self.labels}"
            )
        if count > 1:
            indices = [i for i, l in enumerate(self.labels) if l == label]
            raise ValueError(f"Ambiguous label {label!r} identifies indices {indices}")
        return self.signals[self.labels.index(label)]

    @property
    def patient(self) -> Patient:
        """
        Parsed object representation of the local patient identification.

        See :class:`Patient` for information on its attributes.
        """
        return Patient._from_str(self.local_patient_identification)

    @patient.setter
    def patient(self, patient: Patient) -> None:
        self.local_patient_identification = patient._to_str()

    @property
    def recording(self) -> Recording:
        """
        Parsed object representation of the local recording identification.

        See :class:`Recording` for information on its attributes.
        """
        return Recording._from_str(self.local_recording_identification)

    @recording.setter
    def recording(self, recording: Recording) -> None:
        self._set_startdate_with_recording(recording)
        self.local_recording_identification = recording._to_str()

    @property
    def startdate(self) -> datetime.date:
        """
        Recording startdate.

        If the :attr:`local_recording_identification` conforms to the EDF+ standard, the
        startdate provided there is used. If not, this falls back to the legacy
        :attr:`startdate` field. If both differ, a warning is issued and the EDF+ field
        is preferred. Raises an `AnonymizedDateError` if the EDF+ field is anonymized
        (i.e., begins with `Startdate X`).
        """
        legacy_startdate = decode_date(self._startdate)
        with contextlib.suppress(Exception):
            if legacy_startdate != self.recording.startdate:
                warnings.warn(
                    f"Different values in startdate fields: {legacy_startdate}, {self.recording.startdate}"
                )
        try:
            return self.recording.startdate
        except AnonymizedDateError:
            raise
        except ValueError:
            return legacy_startdate

    @startdate.setter
    def startdate(self, startdate: datetime.date) -> None:
        self._startdate = encode_date(startdate)
        try:
            self.recording.startdate  # noqa: B018
        except AnonymizedDateError:
            pass
        except Exception:
            return
        recording_subfields = self.local_recording_identification.split()
        recording_subfields[1] = _encode_edfplus_date(startdate)
        self.local_recording_identification = " ".join(recording_subfields)

    @property
    def _subsecond_offset(self) -> float:
        try:
            timekeeping_raw = self._timekeeping_signal.digital.tobytes()
            first_data_record = timekeeping_raw[: timekeeping_raw.find(b"\x00") + 1]
            return _EdfAnnotationsDataRecord.from_bytes(first_data_record).tals[0].onset
        except StopIteration:
            return 0

    @property
    def starttime(self) -> datetime.time:
        """
        Recording starttime.

        In EDF+ files, microsecond accuracy is supported.
        """
        subsecond_offset = self._subsecond_offset
        try:
            return decode_time(self._starttime).replace(
                microsecond=round(subsecond_offset * 1000000)
            )
        except ValueError as e:
            raise ValueError(
                f"Subsecond offset in first annotation must be 0.X, is {subsecond_offset}"
            ) from e

    @starttime.setter
    def starttime(self, starttime: datetime.time) -> None:
        onset_change = starttime.microsecond / 1000000 - self._subsecond_offset
        self._starttime = encode_time(starttime.replace(microsecond=0))
        if starttime.microsecond != self.starttime.microsecond:
            for annotation_signal in self._annotation_signals:
                bytes_per_sample = annotation_signal._bytes_per_sample
                data_records = []
                for data_record in annotation_signal.digital.reshape(
                    -1, annotation_signal._bytes_per_data_record
                ):
                    ann_dr = _EdfAnnotationsDataRecord.from_bytes(data_record.tobytes())
                    for tal in ann_dr.tals:
                        tal.onset = round(tal.onset + onset_change, 12)
                    data_records.append(ann_dr.to_bytes())
                maxlen = max(len(data_record) for data_record in data_records)
                maxlen = math.ceil(maxlen / bytes_per_sample) * bytes_per_sample
                raw = b"".join(dr.ljust(maxlen, b"\x00") for dr in data_records)
                annotation_signal._set_samples_per_data_record(
                    maxlen // bytes_per_sample
                )
                annotation_signal._sampling_frequency = (
                    maxlen // bytes_per_sample * self.data_record_duration
                )
                annotation_signal._digital = np.frombuffer(raw, dtype=np.uint8)  # type: ignore[assignment]

    def _set_startdate_with_recording(self, recording: Recording) -> None:
        try:
            startdate = recording.startdate
        except AnonymizedDateError:
            startdate = datetime.date(1985, 1, 1)
        self._startdate = encode_date(startdate)

    @property
    def bytes_in_header_record(self) -> int:
        """Number of bytes in the header record."""
        return int(decode_str(self._bytes_in_header_record))

    @property
    def reserved(self) -> str:
        """`"EDF+C"` for an EDF+C file, else `""`."""
        return decode_str(self._reserved)

    @property
    def num_data_records(self) -> int:
        """Number of data records in the recording."""
        return int(decode_str(self._num_data_records))

    @property
    def data_record_duration(self) -> float:
        """Duration of each data record in seconds."""
        return decode_float(self._data_record_duration)

    def update_data_record_duration(
        self,
        data_record_duration: float,
        method: Literal["strict", "pad", "truncate"] = "strict",
    ) -> None:
        """
        Update the data record duration.

        This operation will fail if the new duration is incompatible with the current
        sampling frequencies.

        Parameters
        ----------
        data_record_duration : float
            The new data record duration in seconds.
        method : `{"strict", "pad", "truncate"}`, default: `"strict"`
            How to handle the case where the new duration does not divide the Edf
            duration evenly

            - "strict": Raise a ValueError
            - "pad": Pad the data with zeros to the next compatible duration. If zero
              is outside the physical range, data is padded with the physical minimum.
            - "truncate": Truncate the data to the previous compatible duration (might
              lead to loss of data)
        """
        if data_record_duration == self.data_record_duration:
            return
        if data_record_duration <= 0:
            raise ValueError(
                f"Data record duration must be positive, got {data_record_duration}"
            )
        if not self.signals:
            raise ValueError(
                "Data record duration must be zero for annotation-only files"
            )
        for signal in self.signals:
            spr = signal.sampling_frequency * data_record_duration
            if spr % 1:
                raise ValueError(
                    f"Cannot set data record duration to {data_record_duration}: Incompatible sampling frequency {signal.sampling_frequency} Hz"
                )

        num_data_records = self._pad_or_truncate_signals(data_record_duration, method)
        self._update_record_duration_in_annotation_signals(
            data_record_duration, num_data_records
        )
        self._set_data_record_duration(data_record_duration)
        self._set_num_data_records(num_data_records)

    @property
    def _total_num_signals(self) -> int:
        return int(decode_str(self._num_signals))

    @property
    def num_signals(self) -> int:
        """Return the number of signals, excluding annotation signals for EDF+."""
        return len(self.signals)

    def _pad_or_truncate_signals(
        self, data_record_duration: float, method: Literal["strict", "pad", "truncate"]
    ) -> int:
        if method == "pad":
            new_duration = (
                ceil(self.duration / data_record_duration) * data_record_duration
            )
            self._pad_or_truncate_data(new_duration)
            return round(new_duration / data_record_duration)
        if method == "truncate":
            new_duration = (
                floor(self.duration / data_record_duration) * data_record_duration
            )
            self._pad_or_truncate_data(new_duration)
            return round(new_duration / data_record_duration)
        return _calculate_num_data_records(self.duration, data_record_duration)

    def _update_record_duration_in_annotation_signals(
        self, data_record_duration: float, num_data_records: int
    ) -> None:
        signals = list(self._signals)
        for idx, signal in enumerate(self._signals):
            if signal not in self._annotation_signals:
                continue
            annotations = []
            for data_record in signal.digital.reshape(
                (-1, signal._bytes_per_data_record)
            ):
                annot_dr = _EdfAnnotationsDataRecord.from_bytes(data_record.tobytes())
                if signal is self._timekeeping_signal:
                    annotations.extend(annot_dr.annotations[1:])
                else:
                    annotations.extend(annot_dr.annotations)
            signals[idx] = _create_annotations_signal(
                [
                    EdfAnnotation(a.onset - self._subsecond_offset, a.duration, a.text)
                    for a in annotations
                ],
                num_data_records=num_data_records,
                data_record_duration=data_record_duration,
                signal_class=self._signal_class,
                with_timestamps=signal is self._timekeeping_signal,
                subsecond_offset=self._subsecond_offset,
            )
        self._signals = tuple(signals)

    def _pad_or_truncate_data(self, new_duration: float) -> None:
        for signal in self.signals:
            n_samples = round(new_duration * signal.sampling_frequency)
            diff = n_samples - len(signal.digital)
            if diff > 0:
                physical_pad_value = 0.0
                if signal.physical_min > 0 or signal.physical_max < 0:
                    physical_pad_value = signal.physical_min
                signal._set_data(
                    np.pad(signal.data, (0, diff), constant_values=physical_pad_value)
                )
            elif diff < 0:
                signal._set_data(signal.data[:diff])

    def anonymize(self) -> None:
        """
        Anonymize a recording.

        Header fields are modified as follows:
          - local patient identification is set to `X X X X`
          - local recording identification is set to `Startdate X X X X`
          - startdate is set to `01.01.85`
          - starttime is set to `00.00.00`

        For EDF+ files, subsecond starttimes specified via an annotations signal are
        removed.
        """
        self.patient = Patient()
        self.recording = Recording()
        self.starttime = datetime.time(0, 0, 0)

    def drop_signals(self, drop: Iterable[int | str]) -> None:
        """
        Drop signals by index or label.

        _Signal indices (int) and labels (str) can be provided in the same iterable. For
        ambiguous labels, all corresponding signals are dropped. Raises a ValueError if
        at least one of the provided identifiers does not correspond to a signal.

        Parameters
        ----------
        drop : Iterable[int  |  str]
            The signals to drop, identified by index or label.
        """
        if isinstance(drop, str):
            drop = [drop]
        selected: list[_Signal] = []
        dropped: list[int | str] = []
        i = 0
        for signal in self._signals:
            if signal._is_annotation_signal:
                selected.append(signal)
                continue
            if i in drop or signal.label in drop:
                dropped.append(i)
                dropped.append(signal.label)
            else:
                selected.append(signal)
            i += 1
        if not_dropped := set(drop) - set(dropped):
            raise ValueError(f"No signal found with index/label {not_dropped}")
        self._signals = tuple(selected)
        self._set_bytes_in_header_record(256 * (len(selected) + 1))
        self._set_num_signals(len(selected))

    def append_signals(self, new_signals: _Signal | Iterable[_Signal]) -> None:
        """
        Append one or more signal(s) to the Edf recording.

        Every signal must be compatible with the current `data_record_duration` and all
        signal durations must match the overall recording duration. For recordings
        containing EDF+ annotation signals, the new signals are inserted after the last
        ordinary (i.e. non-annotation) signal.

        Parameters
        ----------
        new_signals : EdfSignal | Iterable[EdfSignal]
            The signal(s) to add.
        """
        if isinstance(new_signals, (EdfSignal, BdfSignal)):
            new_signals = [new_signals]  # type: ignore[list-item]
        last_ordinary_index = 0
        for i, signal in enumerate(self._signals):
            if not signal._is_annotation_signal:
                last_ordinary_index = i
        self._set_signals(
            [
                *self._signals[: last_ordinary_index + 1],
                *new_signals,
                *self._signals[last_ordinary_index + 1 :],
            ]
        )

    @property
    def _annotation_signals(self) -> Iterable[_Signal]:
        return (s for s in self._signals if s._is_annotation_signal)

    @property
    def _timekeeping_signal(self) -> _Signal:
        return next(iter(self._annotation_signals))

    @property
    def duration(self) -> float:
        """Recording duration in seconds."""
        return self.num_data_records * self.data_record_duration

    def get_annotations(
        self, start_second: float | None = None, stop_second: float | None = None
    ) -> tuple[EdfAnnotation, ...]:
        """
        All annotations defined (starting) in a specified time region of the Edf, sorted chronologically.

        Does not include timekeeping annotations. Will not include annotations that start within the specified time region but are
        defined in data records outside of that window.

        Parameters
        ----------
        start_second : float, optional
            The start of the time region in seconds from recording start. If not provided, the start of the recording is used.
        stop_second : float, optional
            The end of the time region in seconds. If not provided, the end of the recording is used.
        """
        annotations: list[EdfAnnotation] = []
        start_second_defined = start_second is not None
        stop_second_defined = stop_second is not None
        start_second = start_second or 0
        stop_second = stop_second or self.duration
        for i, signal in enumerate(self._annotation_signals):
            if self.duration == 0 or signal.sampling_frequency == 0:
                digital_slice = signal.digital
            else:
                # Make sure to read full data records
                start_record = start_second // self.data_record_duration
                adjusted_start = start_record * self.data_record_duration
                last_record = ceil(
                    stop_second / self.data_record_duration - 1e-12
                )  # avoid floating point errors
                adjusted_stop = last_record * self.data_record_duration
                digital_slice = signal.get_digital_slice(adjusted_start, adjusted_stop)
            for data_record in digital_slice.reshape(
                (-1, signal._bytes_per_data_record)
            ):
                annot_dr = _EdfAnnotationsDataRecord.from_bytes(data_record.tobytes())
                if i == 0:
                    # from https://www.edfplus.info/specs/edfplus.html#timekeeping:
                    # The first annotation of the first 'EDF Annotations' signal in each
                    # data record is empty, but its timestamp specifies how many seconds
                    # after the file startdate/time that data record starts.
                    annotations.extend(annot_dr.annotations[1:])
                else:
                    annotations.extend(annot_dr.annotations)
        subsecond_offset = self._subsecond_offset
        annotations = [
            EdfAnnotation(
                round(ann.onset - subsecond_offset, 12), ann.duration, ann.text
            )
            for ann in annotations
        ]
        annotations = sorted(annotations)
        if start_second_defined:
            while annotations and annotations[0].onset < start_second:
                annotations.pop(0)
        if stop_second_defined:
            while annotations and annotations[-1].onset >= stop_second:
                annotations.pop()
        return tuple(annotations)

    @property
    def annotations(self) -> tuple[EdfAnnotation, ...]:
        """
        All annotations contained in the Edf, sorted chronologically.

        Does not include timekeeping annotations.
        """
        return self.get_annotations()

    def drop_annotations(self, text: str) -> None:
        """
        Drop annotations with a given text.

        Parameters
        ----------
        text : str
            All annotations whose text exactly matches this parameter are removed.
        """
        for signal in self._annotation_signals:
            for data_record in signal.digital.reshape(
                (-1, signal._bytes_per_data_record)
            ):
                annotations = _EdfAnnotationsDataRecord.from_bytes(
                    data_record.tobytes()
                )
                annotations.drop_annotations_with_text(text)
                data_record[:] = np.frombuffer(
                    annotations.to_bytes().ljust(len(data_record), b"\x00"),
                    dtype=np.uint8,
                )

    def set_annotations(self, annotations: Iterable[EdfAnnotation]) -> None:
        """
        Overwrite all annotations with new ones.

        This removes all existing annotation signals and adds a new one as the last
        signal in the file.

        Parameters
        ----------
        annotations : Iterable[EdfAnnotation]
            The annotations to set.
        """
        new_annotation_signal = _create_annotations_signal(
            annotations,
            num_data_records=self.num_data_records,
            data_record_duration=self.data_record_duration,
            signal_class=self._signal_class,
            with_timestamps=True,
            subsecond_offset=self.starttime.microsecond / 1_000_000,
        )
        self._set_signals((*self.signals, new_annotation_signal))

    def add_annotations(self, annotations: Iterable[EdfAnnotation]) -> None:
        """
        Add annotations to the Edf.

        This removes all existing annotation signals and adds a new one containing the
        old and new annotations as the last signal in the file.

        Parameters
        ----------
        annotations : Iterable[EdfAnnotation]
            The annotations to add.
        """
        self.set_annotations(self.annotations + tuple(annotations))

    def to_bytes(self) -> bytes:
        """
        Convert an Edf to a `bytes` object.

        Returns
        -------
        bytes
            The binary representation of the Edf object (i.e., what a file created with
            `Edf.write` would contain).
        """
        stream = io.BytesIO()
        self.write(stream)
        stream.seek(0)
        return stream.read()

    def slice_between_seconds(
        self,
        start: float,
        stop: float,
        *,
        keep_all_annotations: bool = False,
    ) -> None:
        """
        Slice to the interval between two times.

        The sample point corresponding to `stop` is excluded. `start` and `stop` are
        given in seconds from recording start and have to correspond exactly to a sample
        time in all non-annotation signals.

        Parameters
        ----------
        start : float
            Start time in seconds from recording start.
        stop : float
            Stop time in seconds from recording start.
        keep_all_annotations : bool, default: False
            If set to `True`, annotations outside the selected time interval are kept.
        """
        signals: list[_Signal] = []
        self._verify_seconds_inside_recording_time(start)
        self._verify_seconds_inside_recording_time(stop)
        self._verify_seconds_coincide_with_sample_time(start)
        self._verify_seconds_coincide_with_sample_time(stop)
        self._set_num_data_records(
            _calculate_num_data_records(stop - start, self.data_record_duration)
        )
        for signal in self._signals:
            if signal._is_annotation_signal:
                signals.append(
                    self._slice_annotations_signal(
                        signal,
                        start=start,
                        stop=stop,
                        keep_all_annotations=keep_all_annotations,
                    )
                )
            else:
                signal._digital = signal.get_digital_slice(start, stop)
                signals.append(signal)
        self._set_signals(signals)
        self._shift_startdatetime(int(start))

    def slice_between_annotations(
        self,
        start_text: str,
        stop_text: str,
        *,
        keep_all_annotations: bool = False,
    ) -> None:
        """
        Slice to the interval between two EDF+ annotations.

        The sample point corresponding to the onset of the annotation identified by
        `stop_text` is excluded. `start_text` and `stop_text` each have to uniquely
        identify a single annotation, whose onset corresponds exactly to a sample time
        in all non-annotation signals.

        Parameters
        ----------
        start_text : str
            Text identifying the start annotation.
        stop_text : str
            Text identifying the stop annotation.
        keep_all_annotations : bool, default: False
            If set to `True`, annotations outside the selected time interval are kept.
        """
        self.slice_between_seconds(
            self._get_annotation_by_text(start_text).onset,
            self._get_annotation_by_text(stop_text).onset,
            keep_all_annotations=keep_all_annotations,
        )

    def _get_annotation_by_text(self, text: str) -> EdfAnnotation:
        matches = []
        for annotation in self.annotations:
            if annotation.text == text:
                matches.append(annotation)
        if len(matches) == 1:
            return matches[0]
        if len(matches) > 1:
            raise ValueError(
                f"Ambiguous annotation text {text!r}, found {len(matches)} matches"
            )
        raise ValueError(f"No annotation found with text {text!r}")

    def _verify_seconds_inside_recording_time(self, seconds: float) -> None:
        if not 0 <= seconds <= self.duration:
            raise ValueError(
                f"{seconds} is an invalid slice time for recording duration {self.duration}"
            )

    def _verify_seconds_coincide_with_sample_time(self, seconds: float) -> None:
        for i, signal in enumerate(self.signals):
            index = seconds * signal.sampling_frequency
            if index != int(index):
                raise ValueError(
                    f"{seconds}s is not a sample time of signal {i} ({signal.label}) with fs={signal.sampling_frequency}Hz"
                )

    def _shift_startdatetime(self, seconds: float) -> None:
        timedelta = datetime.timedelta(seconds=seconds)
        try:
            startdate = self.startdate
            startdate_anonymized = False
        except AnonymizedDateError:
            startdate = datetime.date.fromtimestamp(0)
            startdate_anonymized = True
        startdatetime = datetime.datetime.combine(startdate, self.starttime)
        startdatetime += timedelta
        if not startdate_anonymized:
            self.startdate = startdatetime.date()
        self.starttime = startdatetime.time()

    def copy(self) -> Self:
        """
        Create a deep copy of the Edf.

        Returns
        -------
        Edf
            The copied Edf object.
        """
        return copy.deepcopy(self)

    def _slice_annotations_signal(
        self,
        signal: _Signal,
        *,
        start: float,
        stop: float,
        keep_all_annotations: bool,
    ) -> _Signal:
        is_timekeeping_signal = signal == self._timekeeping_signal
        annotations: list[EdfAnnotation] = []
        for data_record in signal.digital.reshape((-1, signal._bytes_per_data_record)):
            annot_dr = _EdfAnnotationsDataRecord.from_bytes(data_record.tobytes())
            if is_timekeeping_signal:
                annotations.extend(annot_dr.annotations[1:])
            else:
                annotations.extend(annot_dr.annotations)
        annotations = [
            EdfAnnotation(round(a.onset - start, 12), a.duration, a.text)
            for a in annotations
            if keep_all_annotations or start <= a.onset < stop
        ]
        return _create_annotations_signal(
            annotations,
            num_data_records=self.num_data_records,
            data_record_duration=self.data_record_duration,
            signal_class=self._signal_class,
            with_timestamps=is_timekeeping_signal,
            subsecond_offset=self._subsecond_offset + start - int(start),
        )


class Edf(_Base[EdfSignal]):
    """Python representation of an EDF file.

    EDF header fields are exposed as properties with appropriate data types (i.e.,
    string, numeric, date, or time objects). Fields that might break the file on
    modification (i.e., `version`, `bytes_in_header_record`, `reserved`,
    `num_data_records`, `data_record_duration`, and `num_signals`) can not be set after
    instantiation.

    Note that the startdate has to be set via the parameter `recording`.

    For writing an EDF file with a non-integer seconds duration, currently an
    appropriate value for `data_record_duration` has to be provided manually.

    Parameters
    ----------
    signals : Sequence[EdfSignal]
        The (non-annotation) signals to be contained in the EDF file.
    patient : Patient | None, default: None
        The "local patient identification", containing patient code, sex, birthdate,
        name, and optional additional fields. If `None`, the field is set to `X X X X`
        in accordance with EDF+ specs.
    recording : Recording | None, default: None
        The "local recording identification", containing recording startdate, hospital
        administration code, investigator/technical code, equipment code, and optional
        additional fields. If `None`, the field is set to `Startdate X X X X` in
        accordance with EDF+ specs.
    starttime : datetime.time | None, default: None
        The starttime of the recording. If `None`, `00.00.00` is used. If `starttime`
        contains microseconds, an EDF+C file is created.
    data_record_duration : float | None, default: None
        The duration of each data record in seconds. If `None`, an appropriate value is
        chosen automatically.
    annotations : Iterable[EdfAnnotation] | None, default: None
        The annotations, consisting of onset, duration (optional), and text. If not
        `None`, an EDF+C file is created.
    """

    _signal_class = EdfSignal
    _fmt = "EDF"

    def _set_version(self) -> None:
        self._version = encode_int(0, 8)

    @property
    def version(self) -> int:
        """EDF version, always `0`."""
        return int(decode_str(self._version))


class Bdf(_Base[BdfSignal]):
    """Python representation of a BDF file.

    See :class:`Edf` for information on the header fields and their types.

    .. note::
        BDF uses 24-bit integers (compared to 16-bit for EDF) for the digital values.
        The default for ``digital_range`` (and the supported depth) thus differs.
    """

    _signal_class = BdfSignal
    _fmt = "BDF"

    def _set_version(self) -> None:
        self._version = b"\xffBIOSEMI"

    @property
    def version(self) -> str:
        """The BDF version, always `�BIOSEMI`."""
        return decode_str(self._version)


def _calculate_num_data_records(
    signal_duration: float,
    data_record_duration: float,
) -> int:
    if data_record_duration < 0:
        raise ValueError(
            f"data_record_duration must be positive, got {data_record_duration}"
        )
    for f in (lambda x: x, lambda x: Decimal(str(x))):
        required_num_data_records = f(signal_duration) / f(data_record_duration)
        if required_num_data_records == int(required_num_data_records):
            return int(required_num_data_records)
    raise ValueError(
        f"_Signal duration of {signal_duration}s is not exactly divisible by data_record_duration of {data_record_duration}s"
    )


def _calculate_data_record_duration(signals: Sequence[_Signal]) -> float:
    fs = (Fraction(s.sampling_frequency).limit_denominator(99999999) for s in signals)
    return math.lcm(*(fs_.denominator for fs_ in fs))


_T = TypeVar("_T", bound=Union[Edf, Bdf])


def _read_file(
    file: Path
    | str
    | io.BufferedReader
    | io.BytesIO
    | bytes
    | tempfile.SpooledTemporaryFile[bytes],
    *,
    lazy_load_data: bool,
    header_encoding: str,
    class_: type[_T],
) -> _T:
    if isinstance(file, str):
        file = Path(file)
    if isinstance(file, bytes):
        file = io.BytesIO(file)

    rec = object.__new__(class_)
    if isinstance(file, Path):
        with file.expanduser().open("rb") as f:
            rec._read_header(f, header_encoding)
    else:
        rec._read_header(file, header_encoding)
    rec._load_data(file, lazy_load_data=lazy_load_data)
    return rec


def read_edf(
    edf_file: Path
    | str
    | io.BufferedReader
    | io.BytesIO
    | bytes
    | tempfile.SpooledTemporaryFile[bytes],
    lazy_load_data: bool | Literal["auto"] = "auto",
    *,
    header_encoding: str = "ascii",
) -> Edf:
    """
    Read an EDF file into an :class:`Edf` object.

    If a file-like object is passed, its stream position is moved to EOF.

    Parameters
    ----------
    edf_file : Path | str | io.BufferedReader | io.BytesIO
        The file location (path object or string) or file-like object to read from.
    lazy_load_data : bool | {"auto"}, default: "auto"
        If `True`, the raw signal data is not loaded into memory until it is accessed. If `False`,
        the data is loaded immediately. If `"auto"`, the data is loaded lazily if
        the specified edf_file represents a local path and the extension is not ".bdf"
        and eagerly otherwise.
    header_encoding : str, default: "ascii"
        The character encoding to use when reading header fields.

    Returns
    -------
    Edf
        The resulting :class:`Edf` object.
    """
    if lazy_load_data == "auto":
        lazy_load_data = isinstance(edf_file, (Path, str))
    return _read_file(
        edf_file,
        lazy_load_data=lazy_load_data,
        header_encoding=header_encoding,
        class_=Edf,
    )


def read_bdf(
    bdf_file: Path
    | str
    | io.BufferedReader
    | io.BytesIO
    | bytes
    | tempfile.SpooledTemporaryFile[bytes],
    *,
    header_encoding: str = "ascii",
) -> Bdf:
    """
    Read a BDF file into a :class:`Bdf` object.

    If a file-like object is passed, its stream position is moved to EOF.

    Parameters
    ----------
    bdf_file : Path | str | io.BufferedReader | io.BytesIO
        The file location (path object or string) or file-like object to read from.
    header_encoding : str, default: "ascii"
        The character encoding to use when reading header fields.

    Returns
    -------
    Bdf
        The resulting :class:`Bdf` object.
    """
    return _read_file(
        bdf_file,
        lazy_load_data=False,
        header_encoding=header_encoding,
        class_=Bdf,
    )