class Signal(object):
    """
    The *Signal* represents a channel described by it's samples and timestamps.
    It can perform arithmetic operations against other *Signal* or numeric types.
    The operations are computed in respect to the timestamps (time correct).
    The non-float signals are not interpolated, instead the last value relative
    to the current timestamp is used.
    *samples*, *timestamps* and *name* are mandatory arguments.

    Parameters
    ----------
    samples : numpy.array | list | tuple
        signal samples
    timestamps : numpy.array | list | tuple
        signal timestamps
    unit : str
        signal unit
    name : str
        signal name
    conversion : dict | channel conversion block
        dict that contains extra conversion information about the signal ,
        default *None*
    comment : str
        signal comment, default ''
    raw : bool
        signal samples are raw values, with no physical conversion applied
    master_metadata : list
        master name and sync type
    display_names : dict
        display names used by mdf version 3
    attachment : bytes, name
        channel attachment and name from MDF version 4
    source : Source
        source information named tuple
    bit_count : int
        bit count; useful for integer channels
    stream_sync : bool
        the channel is a synchronisation for the attachment stream (mdf v4 only)
    invalidation_bits : numpy.array | None
        channel invalidation bits, default *None*
    encoding : str | None
        encoding for string signals; default *None*

    """

    def __init__(
        self,
        samples: ArrayLike | None = None,
        timestamps: ArrayLike | None = None,
        unit: str = "",
        name: str = "",
        conversion: dict[str, Any] | ChannelConversionType | None = None,
        comment: str = "",
        raw: bool = True,
        master_metadata: tuple[str, SyncType] | None = None,
        display_names: dict[str, str] | None = None,
        attachment: tuple[bytes, str | None, str | None] | None = None,
        source: SourceType | None = None,
        bit_count: int | None = None,
        stream_sync: bool = False,
        invalidation_bits: ArrayLike | None = None,
        encoding: str | None = None,
        group_index: int = -1,
        channel_index: int = -1,
    ) -> None:

        if samples is None or timestamps is None or not name:
            message = (
                '"samples", "timestamps" and "name" are mandatory '
                f"for Signal class __init__: samples={samples}\n"
                f"timestamps={timestamps}\nname={name}"
            )
            raise MdfException(message)
        else:
            if not isinstance(samples, np.ndarray):
                samples = np.array(samples)
                if samples.dtype.kind == "U":
                    if encoding is None:
                        encodings = ["utf-8", "latin-1"]
                    else:
                        encodings = [encoding, "utf-8", "latin-1"]
                    for encoding in encodings:
                        try:
                            samples = encode(samples, encoding)
                            break
                        except:
                            continue
                    else:
                        samples = encode(samples, encodings[0], errors="ignore")

            if not isinstance(timestamps, np.ndarray):
                timestamps = np.array(timestamps, dtype=np.float64)
            if samples.shape[0] != timestamps.shape[0]:
                message = "{} samples and timestamps length mismatch ({} vs {})"
                message = message.format(name, samples.shape[0], timestamps.shape[0])
                logger.exception(message)
                raise MdfException(message)

            self.samples = samples
            self.timestamps = timestamps
            self.unit = unit
            self.name = name
            self.comment = comment
            self._plot_axis = None
            self.raw = raw
            self.master_metadata = master_metadata
            self.display_names = display_names or {}
            self.attachment = attachment
            self.encoding = encoding
            self.group_index = group_index
            self.channel_index = channel_index

            if source:
                if not isinstance(source, Source):
                    source = Source.from_source(source)
            self.source = source

            if bit_count is None:
                self.bit_count = samples.dtype.itemsize * 8
            else:
                self.bit_count = bit_count

            self.stream_sync = stream_sync

            if invalidation_bits is not None:
                if not isinstance(invalidation_bits, np.ndarray):
                    invalidation_bits = np.array(invalidation_bits)
                if invalidation_bits.shape[0] != samples.shape[0]:
                    message = (
                        "{} samples and invalidation bits length mismatch ({} vs {})"
                    )
                    message = message.format(
                        name, samples.shape[0], invalidation_bits.shape[0]
                    )
                    logger.exception(message)
                    raise MdfException(message)
            self.invalidation_bits = invalidation_bits

            if conversion:
                if not isinstance(
                    conversion, (v4b.ChannelConversion, v3b.ChannelConversion)
                ):
                    conversion = from_dict(conversion)

            self.conversion = conversion

    def __repr__(self):
        return f"""<Signal {self.name}:
\tsamples={self.samples}
\ttimestamps={self.timestamps}
\tinvalidation_bits={self.invalidation_bits}
\tunit="{self.unit}"
\tconversion={self.conversion}
\tsource={self.source}
\tcomment="{self.comment}"
\tmastermeta="{self.master_metadata}"
\traw={self.raw}
\tdisplay_names={self.display_names}
\tattachment={self.attachment}>
"""

    def plot(self, validate: bool = True, index_only: bool = False) -> None:
        """plot Signal samples. Pyqtgraph is used if it is available; in this
        case see the GUI plot documentation to see the available commands

        Parameters
        ----------
        validate (True): bool
            apply the invalidation bits

        index_only (False) : bool
            use index based X axis. This can be useful if the master (usually
            time based) is corrupted with NaN, inf or if it is not strictly
            increasing

        """
        try:

            from .gui.plot import plot

            plot(self, validate=True, index_only=False)
            return

        except:
            try:
                import matplotlib.pyplot as plt
                from matplotlib.widgets import Slider
            except ImportError:
                logging.warning("Signal plotting requires pyqtgraph or matplotlib")
                return

        if len(self.samples.shape) <= 1 and self.samples.dtype.names is None:
            fig = plt.figure()
            fig.canvas.set_window_title(self.name)
            fig.text(
                0.95,
                0.05,
                f"asammdf {__version__}",
                fontsize=8,
                color="red",
                ha="right",
                va="top",
                alpha=0.5,
            )

            name = self.name

            if self.comment:
                comment = self.comment.replace("$", "")
                comment = extract_cncomment_xml(comment)
                comment = fill(comment, 120).replace("\\n", " ")

                title = f"{name}\n({comment})"
                plt.title(title)
            else:
                plt.title(name)
            try:
                if not self.master_metadata:
                    plt.xlabel("Time [s]")
                    plt.ylabel(f"[{self.unit}]")
                    plt.plot(self.timestamps, self.samples, "b")
                    plt.plot(self.timestamps, self.samples, "b.")
                    plt.grid(True)
                    plt.show()
                else:
                    master_name, sync_type = self.master_metadata
                    if sync_type in (0, 1):
                        plt.xlabel(f"{master_name} [s]")
                    elif sync_type == 2:
                        plt.xlabel(f"{master_name} [deg]")
                    elif sync_type == 3:
                        plt.xlabel(f"{master_name} [m]")
                    elif sync_type == 4:
                        plt.xlabel(f"{master_name} [index]")
                    plt.ylabel(f"[{self.unit}]")
                    plt.plot(self.timestamps, self.samples, "b")
                    plt.plot(self.timestamps, self.samples, "b.")
                    plt.grid(True)
                    plt.show()
            except ValueError:
                plt.close(fig)
        else:
            try:
                names = self.samples.dtype.names
                if self.samples.dtype.names is None or len(names) == 1:

                    if names:
                        samples = self.samples[names[0]]
                    else:
                        samples = self.samples

                    shape = samples.shape[1:]

                    fig = plt.figure()
                    fig.canvas.set_window_title(self.name)
                    fig.text(
                        0.95,
                        0.05,
                        f"asammdf {__version__}",
                        fontsize=8,
                        color="red",
                        ha="right",
                        va="top",
                        alpha=0.5,
                    )

                    if self.comment:
                        comment = self.comment.replace("$", "")
                        plt.title(f"{self.name}\n({comment})")
                    else:
                        plt.title(self.name)

                    ax = fig.add_subplot(111, projection="3d")

                    # Grab some test data.
                    X = np.array(range(shape[1]))
                    Y = np.array(range(shape[0]))
                    X, Y = np.meshgrid(X, Y)

                    Z = samples[0]

                    # Plot a basic wireframe.
                    self._plot_axis = ax.plot_wireframe(X, Y, Z, rstride=1, cstride=1)

                    # Place Sliders on Graph
                    ax_a = plt.axes([0.25, 0.1, 0.65, 0.03])

                    # Create Sliders & Determine Range
                    sa = Slider(
                        ax_a,
                        "Time [s]",
                        self.timestamps[0],
                        self.timestamps[-1],
                        valinit=self.timestamps[0],
                    )

                    def update(val):
                        self._plot_axis.remove()
                        idx = np.searchsorted(self.timestamps, sa.val, side="right")
                        Z = samples[idx - 1]
                        self._plot_axis = ax.plot_wireframe(
                            X, Y, Z, rstride=1, cstride=1
                        )
                        fig.canvas.draw_idle()

                    sa.on_changed(update)

                    plt.show()

                else:
                    fig = plt.figure()
                    fig.canvas.set_window_title(self.name)
                    fig.text(
                        0.95,
                        0.05,
                        f"asammdf {__version__}",
                        fontsize=8,
                        color="red",
                        ha="right",
                        va="top",
                        alpha=0.5,
                    )

                    if self.comment:
                        comment = self.comment.replace("$", "")
                        plt.title(f"{self.name}\n({comment})")
                    else:
                        plt.title(self.name)

                    ax = fig.add_subplot(111, projection="3d")

                    samples = self.samples[names[0]]
                    axis1 = self.samples[names[1]]
                    axis2 = self.samples[names[2]]

                    # Grab some test data.
                    X, Y = np.meshgrid(axis2[0], axis1[0])

                    Z = samples[0]

                    # Plot a basic wireframe.
                    self._plot_axis = ax.plot_wireframe(X, Y, Z, rstride=1, cstride=1)

                    # Place Sliders on Graph
                    ax_a = plt.axes([0.25, 0.1, 0.65, 0.03])

                    # Create Sliders & Determine Range
                    sa = Slider(
                        ax_a,
                        "Time [s]",
                        self.timestamps[0],
                        self.timestamps[-1],
                        valinit=self.timestamps[0],
                    )

                    def update(val):
                        self._plot_axis.remove()
                        idx = np.searchsorted(self.timestamps, sa.val, side="right")
                        Z = samples[idx - 1]
                        X, Y = np.meshgrid(axis2[idx - 1], axis1[idx - 1])
                        self._plot_axis = ax.plot_wireframe(
                            X, Y, Z, rstride=1, cstride=1
                        )
                        fig.canvas.draw_idle()

                    sa.on_changed(update)

                    plt.show()

            except Exception as err:
                print(err)

    def cut(
        self,
        start: float | None = None,
        stop: float | None = None,
        include_ends: bool = True,
        integer_interpolation_mode: IntInterpolationModeType
        | IntegerInterpolation = IntegerInterpolation.REPEAT_PREVIOUS_SAMPLE,
        float_interpolation_mode: FloatInterpolationModeType
        | FloatInterpolation = FloatInterpolation.LINEAR_INTERPOLATION,
    ) -> Signal:
        """
        Cuts the signal according to the *start* and *stop* values, by using
        the insertion indexes in the signal's *time* axis.

        Parameters
        ----------
        start : float
            start timestamp for cutting
        stop : float
            stop timestamp for cutting
        include_ends : bool
            include the *start* and *stop* timestamps after cutting the signal.
            If *start* and *stop* are found in the original timestamps, then
            the new samples will be computed using interpolation. Default *True*

        integer_interpolation_mode : int
            interpolation mode for integer signals; default 0

                * 0 - repeat previous samples
                * 1 - linear interpolation
                * 2 - hybrid interpolation: channels with integer data type (raw values) that have a
                  conversion that outputs float values will use linear interpolation, otherwise
                  the previous sample is used

                .. versionadded:: 6.2.0

        float_interpolation_mode : int
            interpolation mode for float channels; default 1

                * 0 - repeat previous sample
                * 1 - use linear interpolation

                .. versionadded:: 6.2.0

        Returns
        -------
        result : Signal
            new *Signal* cut from the original

        Examples
        --------
        >>> new_sig = old_sig.cut(1.0, 10.5)
        >>> new_sig.timestamps[0], new_sig.timestamps[-1]
        0.98, 10.48

        """

        integer_interpolation_mode = IntegerInterpolation(integer_interpolation_mode)
        float_interpolation_mode = FloatInterpolation(float_interpolation_mode)

        original_start, original_stop = (start, stop)

        if len(self) == 0:
            result = Signal(
                np.array([], dtype=self.samples.dtype),
                np.array([], dtype=self.timestamps.dtype),
                self.unit,
                self.name,
                self.conversion,
                self.comment,
                self.raw,
                self.master_metadata,
                self.display_names,
                self.attachment,
                self.source,
                self.bit_count,
                self.stream_sync,
                encoding=self.encoding,
                group_index=self.group_index,
                channel_index=self.channel_index,
            )

        elif start is None and stop is None:
            # return the channel uncut
            result = Signal(
                self.samples.copy(),
                self.timestamps.copy(),
                self.unit,
                self.name,
                self.conversion,
                self.comment,
                self.raw,
                self.master_metadata,
                self.display_names,
                self.attachment,
                self.source,
                self.bit_count,
                self.stream_sync,
                invalidation_bits=self.invalidation_bits.copy()
                if self.invalidation_bits is not None
                else None,
                encoding=self.encoding,
                group_index=self.group_index,
                channel_index=self.channel_index,
            )

        else:
            if start is None:
                # cut from begining to stop
                if stop < self.timestamps[0]:
                    result = Signal(
                        np.array([], dtype=self.samples.dtype),
                        np.array([], dtype=self.timestamps.dtype),
                        self.unit,
                        self.name,
                        self.conversion,
                        self.comment,
                        self.raw,
                        self.master_metadata,
                        self.display_names,
                        self.attachment,
                        self.source,
                        self.bit_count,
                        self.stream_sync,
                        encoding=self.encoding,
                        group_index=self.group_index,
                        channel_index=self.channel_index,
                    )

                else:
                    stop = np.searchsorted(self.timestamps, stop, side="right")
                    if (
                        include_ends
                        and original_stop not in self.timestamps
                        and original_stop < self.timestamps[-1]
                    ):
                        interpolated = self.interp(
                            [original_stop],
                            integer_interpolation_mode=integer_interpolation_mode,
                            float_interpolation_mode=float_interpolation_mode,
                        )

                        if len(interpolated):
                            samples = np.append(
                                self.samples[:stop], interpolated.samples, axis=0
                            )
                            timestamps = np.append(
                                self.timestamps[:stop], interpolated.timestamps
                            )
                            if self.invalidation_bits is not None:
                                invalidation_bits = np.append(
                                    self.invalidation_bits[:stop],
                                    interpolated.invalidation_bits,
                                )
                            else:
                                invalidation_bits = None
                    else:
                        samples = self.samples[:stop].copy()
                        timestamps = self.timestamps[:stop].copy()
                        if self.invalidation_bits is not None:
                            invalidation_bits = self.invalidation_bits[:stop].copy()
                        else:
                            invalidation_bits = None
                    result = Signal(
                        samples,
                        timestamps,
                        self.unit,
                        self.name,
                        self.conversion,
                        self.comment,
                        self.raw,
                        self.master_metadata,
                        self.display_names,
                        self.attachment,
                        self.source,
                        self.bit_count,
                        self.stream_sync,
                        invalidation_bits=invalidation_bits,
                        encoding=self.encoding,
                        group_index=self.group_index,
                        channel_index=self.channel_index,
                    )

            elif stop is None:
                # cut from start to end
                if start > self.timestamps[-1]:
                    result = Signal(
                        np.array([], dtype=self.samples.dtype),
                        np.array([], dtype=self.timestamps.dtype),
                        self.unit,
                        self.name,
                        self.conversion,
                        self.comment,
                        self.raw,
                        self.master_metadata,
                        self.display_names,
                        self.attachment,
                        self.source,
                        self.bit_count,
                        self.stream_sync,
                        encoding=self.encoding,
                        group_index=self.group_index,
                        channel_index=self.channel_index,
                    )

                else:
                    start = np.searchsorted(self.timestamps, start, side="left")
                    if (
                        include_ends
                        and original_start not in self.timestamps
                        and original_start > self.timestamps[0]
                    ):
                        interpolated = self.interp(
                            [original_start],
                            integer_interpolation_mode=integer_interpolation_mode,
                            float_interpolation_mode=float_interpolation_mode,
                        )
                        if len(interpolated):
                            samples = np.append(
                                interpolated.samples, self.samples[start:], axis=0
                            )
                            timestamps = np.append(
                                interpolated.timestamps, self.timestamps[start:]
                            )
                            if self.invalidation_bits is not None:
                                invalidation_bits = np.append(
                                    interpolated.invalidation_bits,
                                    self.invalidation_bits[start:],
                                )
                            else:
                                invalidation_bits = None
                    else:
                        samples = self.samples[start:].copy()
                        timestamps = self.timestamps[start:].copy()
                        if self.invalidation_bits is not None:
                            invalidation_bits = self.invalidation_bits[start:].copy()
                        else:
                            invalidation_bits = None
                    result = Signal(
                        samples,
                        timestamps,
                        self.unit,
                        self.name,
                        self.conversion,
                        self.comment,
                        self.raw,
                        self.master_metadata,
                        self.display_names,
                        self.attachment,
                        self.source,
                        self.bit_count,
                        self.stream_sync,
                        invalidation_bits=invalidation_bits,
                        encoding=self.encoding,
                        group_index=self.group_index,
                        channel_index=self.channel_index,
                    )

            else:
                # cut between start and stop
                if start > self.timestamps[-1] or stop < self.timestamps[0]:
                    result = Signal(
                        np.array([], dtype=self.samples.dtype),
                        np.array([], dtype=self.timestamps.dtype),
                        self.unit,
                        self.name,
                        self.conversion,
                        self.comment,
                        self.raw,
                        self.master_metadata,
                        self.display_names,
                        self.attachment,
                        self.source,
                        self.bit_count,
                        self.stream_sync,
                        encoding=self.encoding,
                        group_index=self.group_index,
                        channel_index=self.channel_index,
                    )
                else:
                    start = np.searchsorted(self.timestamps, start, side="left")
                    stop = np.searchsorted(self.timestamps, stop, side="right")

                    if start == stop:
                        if include_ends:
                            if original_start == original_stop:
                                ends = np.array(
                                    [original_start], dtype=self.timestamps.dtype
                                )
                            else:
                                ends = np.array(
                                    [original_start, original_stop],
                                    dtype=self.timestamps.dtype,
                                )

                            interpolated = self.interp(
                                ends,
                                integer_interpolation_mode=integer_interpolation_mode,
                                float_interpolation_mode=float_interpolation_mode,
                            )
                            samples = interpolated.samples
                            timestamps = ends
                            invalidation_bits = interpolated.invalidation_bits
                        else:
                            samples = np.array([], dtype=self.samples.dtype)
                            timestamps = np.array([], dtype=self.timestamps.dtype)
                            if self.invalidation_bits is not None:
                                invalidation_bits = np.array([], dtype=bool)
                            else:
                                invalidation_bits = None
                    else:
                        samples = self.samples[start:stop].copy()
                        timestamps = self.timestamps[start:stop].copy()
                        if self.invalidation_bits is not None:
                            invalidation_bits = self.invalidation_bits[
                                start:stop
                            ].copy()
                        else:
                            invalidation_bits = None

                        if (
                            include_ends
                            and original_stop not in self.timestamps
                            and original_stop < self.timestamps[-1]
                        ):
                            interpolated = self.interp(
                                [original_stop],
                                integer_interpolation_mode=integer_interpolation_mode,
                                float_interpolation_mode=float_interpolation_mode,
                            )

                            if len(interpolated):
                                samples = np.append(
                                    samples, interpolated.samples, axis=0
                                )
                                timestamps = np.append(
                                    timestamps, interpolated.timestamps
                                )
                                if invalidation_bits is not None:
                                    invalidation_bits = np.append(
                                        invalidation_bits,
                                        interpolated.invalidation_bits,
                                    )

                        if (
                            include_ends
                            and original_start not in self.timestamps
                            and original_start > self.timestamps[0]
                        ):
                            interpolated = self.interp(
                                [original_start],
                                integer_interpolation_mode=integer_interpolation_mode,
                                float_interpolation_mode=float_interpolation_mode,
                            )

                            if len(interpolated):
                                samples = np.append(
                                    interpolated.samples, samples, axis=0
                                )
                                timestamps = np.append(
                                    interpolated.timestamps, timestamps
                                )

                                if invalidation_bits is not None:
                                    invalidation_bits = np.append(
                                        interpolated.invalidation_bits,
                                        invalidation_bits,
                                    )

                    result = Signal(
                        samples,
                        timestamps,
                        self.unit,
                        self.name,
                        self.conversion,
                        self.comment,
                        self.raw,
                        self.master_metadata,
                        self.display_names,
                        self.attachment,
                        self.source,
                        self.bit_count,
                        self.stream_sync,
                        invalidation_bits=invalidation_bits,
                        encoding=self.encoding,
                        group_index=self.group_index,
                        channel_index=self.channel_index,
                    )

        return result

    def extend(self, other: Signal) -> Signal:
        """extend signal with samples from another signal

        Parameters
        ----------
        other : Signal

        Returns
        -------
        signal : Signal
            new extended *Signal*

        """
        if len(self.timestamps):
            last_stamp = self.timestamps[-1]
        else:
            last_stamp = 0
        if len(other):
            other_first_sample = other.timestamps[0]
            if last_stamp >= other_first_sample:
                timestamps = other.timestamps + last_stamp
            else:
                timestamps = other.timestamps

            if self.invalidation_bits is None and other.invalidation_bits is None:
                invalidation_bits = None
            elif self.invalidation_bits is None and other.invalidation_bits is not None:
                invalidation_bits = np.concatenate(
                    (np.zeros(len(self), dtype=bool), other.invalidation_bits)
                )
            elif self.invalidation_bits is not None and other.invalidation_bits is None:
                invalidation_bits = np.concatenate(
                    (self.invalidation_bits, np.zeros(len(other), dtype=bool))
                )
            else:
                invalidation_bits = np.append(
                    self.invalidation_bits, other.invalidation_bits
                )

            result = Signal(
                np.append(self.samples, other.samples, axis=0),
                np.append(self.timestamps, timestamps),
                self.unit,
                self.name,
                self.conversion,
                self.comment,
                self.raw,
                self.master_metadata,
                self.display_names,
                self.attachment,
                self.source,
                self.bit_count,
                self.stream_sync,
                invalidation_bits=invalidation_bits,
                encoding=self.encoding,
                group_index=self.group_index,
                channel_index=self.channel_index,
            )
        else:
            result = self

        return result

    def interp(
        self,
        new_timestamps: NDArray[Any],
        integer_interpolation_mode: IntInterpolationModeType
        | IntegerInterpolation = IntegerInterpolation.REPEAT_PREVIOUS_SAMPLE,
        float_interpolation_mode: FloatInterpolationModeType
        | FloatInterpolation = FloatInterpolation.LINEAR_INTERPOLATION,
    ) -> Signal:
        """returns a new *Signal* interpolated using the *new_timestamps*

        Parameters
        ----------
        new_timestamps : np.array
            timestamps used for interpolation

        integer_interpolation_mode : int
            interpolation mode for integer signals; default 0

                * 0 - repeat previous samples
                * 1 - linear interpolation
                * 2 - hybrid interpolation: channels with integer data type (raw values) that have a
                  conversion that outputs float values will use linear interpolation, otherwise
                  the previous sample is used

                .. versionadded:: 6.2.0

        float_interpolation_mode : int
            interpolation mode for float channels; default 1

                * 0 - repeat previous sample
                * 1 - use linear interpolation

                .. versionadded:: 6.2.0

        Returns
        -------
        signal : Signal
            new interpolated *Signal*

        """

        integer_interpolation_mode = IntegerInterpolation(integer_interpolation_mode)
        float_interpolation_mode = FloatInterpolation(float_interpolation_mode)

        if not len(self.samples) or not len(new_timestamps):

            return Signal(
                self.samples[:0].copy(),
                self.timestamps[:0].copy(),
                self.unit,
                self.name,
                comment=self.comment,
                conversion=self.conversion,
                raw=self.raw,
                master_metadata=self.master_metadata,
                display_names=self.display_names,
                attachment=self.attachment,
                stream_sync=self.stream_sync,
                invalidation_bits=None,
                encoding=self.encoding,
                group_index=self.group_index,
                channel_index=self.channel_index,
            )
        else:

            # # we need to validate first otherwise we can get false invalid data
            # # if the new timebase and the invaldiation bits are alligned in an
            # # infavorable way
            #
            # if self.invalidation_bits is not None:
            #     signal = self.validate()
            #     has_invalidation = True
            # else:
            #     signal = self
            #     has_invalidation = False

            signal = self
            invalidation_bits = signal.invalidation_bits

            if not len(signal.samples):
                return Signal(
                    self.samples[:0].copy(),
                    self.timestamps[:0].copy(),
                    self.unit,
                    self.name,
                    comment=self.comment,
                    conversion=self.conversion,
                    raw=self.raw,
                    master_metadata=self.master_metadata,
                    display_names=self.display_names,
                    attachment=self.attachment,
                    stream_sync=self.stream_sync,
                    invalidation_bits=None
                    if invalidation_bits is None
                    else np.array([], dtype=bool),
                    encoding=self.encoding,
                    group_index=self.group_index,
                    channel_index=self.channel_index,
                )

            if len(signal.samples.shape) > 1:
                idx = np.searchsorted(signal.timestamps, new_timestamps, side="right")
                idx -= 1
                idx[idx < 0] = 0
                s = signal.samples[idx]
                if invalidation_bits is not None:
                    invalidation_bits = invalidation_bits[idx]
            else:

                kind = signal.samples.dtype.kind

                if kind == "f":

                    if (
                        float_interpolation_mode
                        == FloatInterpolation.REPEAT_PREVIOUS_SAMPLE
                    ):
                        idx = np.searchsorted(
                            signal.timestamps, new_timestamps, side="right"
                        )
                        idx -= 1
                        idx[idx < 0] = 0
                        s = signal.samples[idx]

                        if invalidation_bits is not None:
                            invalidation_bits = invalidation_bits[idx]

                    else:
                        s = np.interp(new_timestamps, signal.timestamps, signal.samples)

                        if invalidation_bits is not None:
                            idx = np.searchsorted(
                                signal.timestamps, new_timestamps, side="right"
                            )
                            idx -= 1
                            idx[idx < 0] = 0
                            invalidation_bits = invalidation_bits[idx]

                elif kind in "ui":
                    if (
                        integer_interpolation_mode
                        == IntegerInterpolation.HYBRID_INTERPOLATION
                    ):
                        if signal.raw and signal.conversion:
                            kind = signal.conversion.convert(
                                signal.samples[:1]
                            ).dtype.kind
                            if kind == "f":
                                integer_interpolation_mode = (
                                    IntegerInterpolation.LINEAR_INTERPOLATION
                                )

                    if (
                        integer_interpolation_mode
                        == IntegerInterpolation.HYBRID_INTERPOLATION
                    ):
                        integer_interpolation_mode = (
                            IntegerInterpolation.REPEAT_PREVIOUS_SAMPLE
                        )

                    if (
                        integer_interpolation_mode
                        == IntegerInterpolation.LINEAR_INTERPOLATION
                    ):
                        s = np.interp(
                            new_timestamps, signal.timestamps, signal.samples
                        ).astype(signal.samples.dtype)

                        if invalidation_bits is not None:
                            idx = np.searchsorted(
                                signal.timestamps, new_timestamps, side="right"
                            )
                            idx -= 1
                            idx[idx < 0] = 0
                            invalidation_bits = invalidation_bits[idx]

                    elif (
                        integer_interpolation_mode
                        == IntegerInterpolation.REPEAT_PREVIOUS_SAMPLE
                    ):
                        idx = np.searchsorted(
                            signal.timestamps, new_timestamps, side="right"
                        )
                        idx -= 1
                        idx[idx < 0] = 0

                        s = signal.samples[idx]

                        if invalidation_bits is not None:
                            invalidation_bits = invalidation_bits[idx]

                else:
                    idx = np.searchsorted(
                        signal.timestamps, new_timestamps, side="right"
                    )
                    idx -= 1
                    idx[idx < 0] = 0
                    s = signal.samples[idx]

                    if invalidation_bits is not None:
                        invalidation_bits = invalidation_bits[idx]

            return Signal(
                s,
                new_timestamps,
                self.unit,
                self.name,
                comment=self.comment,
                conversion=self.conversion,
                source=self.source,
                raw=self.raw,
                master_metadata=self.master_metadata,
                display_names=self.display_names,
                attachment=self.attachment,
                stream_sync=self.stream_sync,
                invalidation_bits=invalidation_bits,
                encoding=self.encoding,
                group_index=self.group_index,
                channel_index=self.channel_index,
            )

    def __apply_func(
        self, other: Signal | NDArray[Any] | None, func_name: str
    ) -> Signal:
        """delegate operations to the *samples* attribute, but in a time
        correct manner by considering the *timestamps*

        """

        if isinstance(other, Signal):
            if len(self) and len(other):
                start = max(self.timestamps[0], other.timestamps[0])
                stop = min(self.timestamps[-1], other.timestamps[-1])
                s1 = self.physical().cut(start, stop)
                s2 = other.physical().cut(start, stop)
            else:
                s1 = self
                s2 = other
            time = np.union1d(s1.timestamps, s2.timestamps)
            s = s1.interp(time).samples
            o = s2.interp(time).samples
            func = getattr(s, func_name)
            conversion = None
            s = func(o)
        elif other is None:
            s = self.samples
            conversion = self.conversion
            time = self.timestamps
        else:
            func = getattr(self.samples, func_name)
            s = func(other)
            conversion = self.conversion
            time = self.timestamps
        return Signal(
            samples=s,
            timestamps=time,
            unit=self.unit,
            name=self.name,
            conversion=conversion,
            raw=self.raw,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
            encoding=self.encoding,
            group_index=self.group_index,
            channel_index=self.channel_index,
        )

    def __pos__(self) -> Signal:
        return self

    def __neg__(self) -> Signal:
        return Signal(
            np.negative(self.samples),
            self.timestamps,
            unit=self.unit,
            name=self.name,
            conversion=self.conversion,
            raw=self.raw,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
            encoding=self.encoding,
        )

    def __round__(self, n: int) -> Signal:
        return Signal(
            np.around(self.samples, n),
            self.timestamps,
            unit=self.unit,
            name=self.name,
            conversion=self.conversion,
            raw=self.raw,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
            encoding=self.encoding,
        )

    def __sub__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__sub__")

    def __isub__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__sub__(other)

    def __rsub__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return -self.__sub__(other)

    def __add__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__add__")

    def __iadd__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__add__(other)

    def __radd__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__add__(other)

    def __truediv__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__truediv__")

    def __itruediv__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__truediv__(other)

    def __rtruediv__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__rtruediv__")

    def __mul__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__mul__")

    def __imul__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__mul__(other)

    def __rmul__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__mul__(other)

    def __floordiv__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__floordiv__")

    def __ifloordiv__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__truediv__(other)

    def __rfloordiv__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return 1 / self.__apply_func(other, "__rfloordiv__")

    def __mod__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__mod__")

    def __pow__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__pow__")

    def __and__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__and__")

    def __or__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__or__")

    def __xor__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__xor__")

    def __invert__(self) -> Signal:
        s = ~self.samples
        time = self.timestamps
        return Signal(
            s,
            time,
            unit=self.unit,
            name=self.name,
            conversion=self.conversion,
            raw=self.raw,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
            encoding=self.encoding,
        )

    def __lshift__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__lshift__")

    def __rshift__(self, other: Signal | NDArray[Any] | None) -> Signal:
        return self.__apply_func(other, "__rshift__")

    def __lt__(self, other: Signal | NDArray[Any] | None) -> bool:
        return self.__apply_func(other, "__lt__")

    def __le__(self, other: Signal | NDArray[Any] | None) -> bool:
        return self.__apply_func(other, "__le__")

    def __gt__(self, other: Signal | NDArray[Any] | None) -> bool:
        return self.__apply_func(other, "__gt__")

    def __ge__(self, other: Signal | NDArray[Any] | None) -> bool:
        return self.__apply_func(other, "__ge__")

    def __eq__(self, other: Signal | NDArray[Any] | None) -> bool:
        return self.__apply_func(other, "__eq__")

    def __ne__(self, other: Signal | NDArray[Any] | None) -> bool:
        return self.__apply_func(other, "__ne__")

    def __iter__(self) -> Iterator[Any]:
        for item in (self.samples, self.timestamps, self.unit, self.name):
            yield item

    def __reversed__(self) -> Iterator[tuple[int, tuple[Any, Any]]]:
        return enumerate(zip(reversed(self.samples), reversed(self.timestamps)))

    def __len__(self) -> int:
        return len(self.samples)

    def __abs__(self) -> Signal:
        return Signal(
            np.fabs(self.samples),
            self.timestamps,
            unit=self.unit,
            name=self.name,
            conversion=self.conversion,
            raw=self.raw,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
        )

    def __getitem__(self, val: int) -> Any:
        return self.samples[val]

    def __setitem__(self, idx: int, val: Any) -> None:
        self.samples[idx] = val

    def astype(self, np_type: DTypeLike) -> Signal:
        """returns new *Signal* with samples of dtype *np_type*

        Parameters
        ----------
        np_type : np.dtype
            new numpy dtye

        Returns
        -------
        signal : Signal
            new *Signal* with the samples of *np_type* dtype

        """
        return Signal(
            self.samples.astype(np_type),
            self.timestamps,
            unit=self.unit,
            name=self.name,
            conversion=self.conversion,
            raw=self.raw,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
            encoding=self.encoding,
        )

    def physical(self) -> Signal:
        """
        get the physical samples values

        Returns
        -------
        phys : Signal
            new *Signal* with physical values

        """

        if not self.raw or self.conversion is None:
            samples = self.samples.copy()
            encoding = None
        else:
            samples = self.conversion.convert(self.samples)
            if samples.dtype.kind == "S":
                encoding = "utf-8" if self.conversion.id == b"##CC" else "latin-1"
            else:
                encoding = None

        return Signal(
            samples,
            self.timestamps.copy(),
            unit=self.unit,
            name=self.name,
            conversion=None,
            raw=False,
            master_metadata=self.master_metadata,
            display_names=self.display_names,
            attachment=self.attachment,
            stream_sync=self.stream_sync,
            invalidation_bits=self.invalidation_bits,
            source=self.source,
            encoding=encoding,
            group_index=self.group_index,
            channel_index=self.channel_index,
        )

    def validate(self, copy: bool = True) -> Signal:
        """appply invalidation bits if they are available for this signal

        Parameters
        ----------
        copy (True) : bool
            return a copy of the result

            .. versionadded:: 5.12.0

        """
        if self.invalidation_bits is None:
            signal = self

        else:
            idx = np.nonzero(~self.invalidation_bits)[0]
            signal = Signal(
                self.samples[idx],
                self.timestamps[idx],
                self.unit,
                self.name,
                self.conversion,
                self.comment,
                self.raw,
                self.master_metadata,
                self.display_names,
                self.attachment,
                self.source,
                self.bit_count,
                self.stream_sync,
                invalidation_bits=None,
                encoding=self.encoding,
                group_index=self.group_index,
                channel_index=self.channel_index,
            )

        if copy:
            signal = signal.copy()

        return signal

    def copy(self) -> Signal:
        """copy all attributes to a new Signal"""
        return Signal(
            self.samples.copy(),
            self.timestamps.copy(),
            self.unit,
            self.name,
            self.conversion,
            self.comment,
            self.raw,
            self.master_metadata,
            self.display_names,
            self.attachment,
            self.source,
            self.bit_count,
            self.stream_sync,
            invalidation_bits=self.invalidation_bits.copy()
            if self.invalidation_bits is not None
            else None,
            encoding=self.encoding,
            group_index=self.group_index,
            channel_index=self.channel_index,
        ) def conversion_transfer(
    conversion: ChannelConversionType, version: int = 3
) -> ChannelConversionType:
    """convert between mdf4 and mdf3 channel conversions

    Parameters
    ----------
    conversion : block
        channel conversion
    version : int
        target mdf version

    Returns
    -------
    conversion : block
        channel conversion for specified version

    """

    if version <= 3:
        if conversion is None:
            conversion = v3b.ChannelConversion(conversion_type=v3c.CONVERSION_TYPE_NONE)
        else:
            conversion_type = conversion["conversion_type"]
            if conversion.id == b"CC":
                pass
            else:
                unit = conversion.unit.strip(" \r\n\t\0").encode("latin-1")

                if conversion_type == v4c.CONVERSION_TYPE_NON:
                    conversion = v3b.ChannelConversion(
                        unit=unit, conversion_type=v3c.CONVERSION_TYPE_NONE
                    )

                elif conversion_type == v4c.CONVERSION_TYPE_LIN:
                    conversion = v3b.ChannelConversion(
                        unit=unit,
                        conversion_type=v3c.CONVERSION_TYPE_LINEAR,
                        a=conversion.a,
                        b=conversion.b,
                    )

                elif conversion_type == v4c.CONVERSION_TYPE_RAT:
                    conversion = v3b.ChannelConversion(
                        unit=unit,
                        conversion_type=v3c.CONVERSION_TYPE_RAT,
                        P1=conversion.P1,
                        P2=conversion.P2,
                        P3=conversion.P3,
                        P4=conversion.P4,
                        P5=conversion.P5,
                        P6=conversion.P6,
                    )

                elif conversion_type == v4c.CONVERSION_TYPE_TAB:
                    conversion_ = {}
                    conversion_["ref_param_nr"] = conversion.val_param_nr // 2
                    for i in range(conversion.val_param_nr // 2):
                        conversion_[f"raw_{i}"] = conversion[f"raw_{i}"]
                        conversion_[f"phys_{i}"] = conversion[f"phys_{i}"]

                    conversion = v3b.ChannelConversion(
                        unit=unit,
                        conversion_type=v3c.CONVERSION_TYPE_TAB,
                        **conversion_,
                    )

                elif conversion_type == v4c.CONVERSION_TYPE_TABI:
                    conversion_ = {}
                    conversion_["ref_param_nr"] = conversion.val_param_nr // 2
                    for i in range(conversion.val_param_nr // 2):
                        conversion_[f"raw_{i}"] = conversion[f"raw_{i}"]
                        conversion_[f"phys_{i}"] = conversion[f"phys_{i}"]

                    conversion = v3b.ChannelConversion(
                        unit=unit,
                        conversion_type=v3c.CONVERSION_TYPE_TABI,
                        **conversion_,
                    )

                elif conversion_type == v4c.CONVERSION_TYPE_ALG:
                    formula = conversion.formula.replace("X", "X1")
                    conversion = v3b.ChannelConversion(
                        formula=formula,
                        unit=unit,
                        conversion_type=v3c.CONVERSION_TYPE_FORMULA,
                    )

                elif conversion_type == v4c.CONVERSION_TYPE_RTAB:
                    nr = (conversion.val_param_nr - 1) // 3
                    kargs = {
                        "ref_param_nr": nr,
                        "conversion_type": v3c.CONVERSION_TYPE_TABI,
                    }

                    for i in range(nr):
                        l_ = conversion[f"lower_{i}"]
                        u_ = conversion[f"upper_{i}"]
                        p_ = conversion[f"phys_{i}"]
                        kargs[f"raw_{i}"] = l_
                        kargs[f"raw_{i}"] = u_ - 0.000_001
                        kargs[f"phys_{i}"] = p_
                        kargs[f"phys_{i}"] = p_

                    conversion = v3b.ChannelConversion(unit=unit, **kargs)

                elif conversion_type == v4c.CONVERSION_TYPE_TABX:
                    nr = conversion.val_param_nr

                    kargs = {
                        "ref_param_nr": nr + 1,
                        "unit": unit,
                        "conversion_type": v3c.CONVERSION_TYPE_RTABX,
                    }
                    for i in range(nr):
                        kargs[f"lower_{i}"] = conversion[f"val_{i}"]
                        kargs[f"upper_{i}"] = conversion[f"val_{i}"]
                        if isinstance(
                            conversion.referenced_blocks[f"text_{i}"],
                            v4b.ChannelConversion,
                        ):
                            kargs[f"text_{i}"] = conversion.referenced_blocks[
                                f"text_{i}"
                            ].name.encode("latin-1")
                        else:
                            kargs[f"text_{i}"] = conversion.referenced_blocks[
                                f"text_{i}"
                            ]

                    new_conversion = v3b.ChannelConversion(**kargs)
                    if isinstance(
                        conversion.referenced_blocks["default_addr"],
                        v4b.ChannelConversion,
                    ):
                        default_addr = conversion.referenced_blocks[
                            f"default_addr"
                        ].name.encode("latin-1")
                    else:
                        default_addr = conversion.referenced_blocks["default_addr"]
                    new_conversion.referenced_blocks["default_addr"] = default_addr

                    conversion = new_conversion

                elif conversion_type == v4c.CONVERSION_TYPE_RTABX:
                    nr = conversion.val_param_nr // 2
                    kargs = {
                        "ref_param_nr": nr + 1,
                        "unit": unit,
                        "conversion_type": v3c.CONVERSION_TYPE_RTABX,
                    }
                    for i in range(nr):
                        kargs[f"lower_{i}"] = conversion[f"lower_{i}"]
                        kargs[f"upper_{i}"] = conversion[f"upper_{i}"]
                        if isinstance(
                            conversion.referenced_blocks[f"text_{i}"],
                            v4b.ChannelConversion,
                        ):
                            kargs[f"text_{i}"] = conversion.referenced_blocks[
                                f"text_{i}"
                            ].name.encode("latin-1")
                        else:
                            kargs[f"text_{i}"] = conversion.referenced_blocks[
                                f"text_{i}"
                            ]

                    new_conversion = v3b.ChannelConversion(**kargs)
                    if isinstance(
                        conversion.referenced_blocks["default_addr"],
                        v4b.ChannelConversion,
                    ):
                        default_addr = conversion.referenced_blocks[
                            f"default_addr"
                        ].name.encode("latin-1")
                    else:
                        default_addr = conversion.referenced_blocks["default_addr"]
                    new_conversion.referenced_blocks["default_addr"] = default_addr

                    conversion = new_conversion

    else:
        if conversion is None or conversion.id == b"##CC":
            pass
        else:
            conversion_type = conversion.conversion_type
            unit = conversion.unit_field.decode("latin-1").strip(" \r\n\t\0")

            if conversion_type == v3c.CONVERSION_TYPE_NONE:
                conversion = v4b.ChannelConversion(
                    conversion_type=v4c.CONVERSION_TYPE_NON
                )

            elif conversion_type == v3c.CONVERSION_TYPE_LINEAR:
                conversion = v4b.ChannelConversion(
                    conversion_type=v4c.CONVERSION_TYPE_LIN,
                    a=conversion.a,
                    b=conversion.b,
                )

            elif conversion_type == v3c.CONVERSION_TYPE_RAT:
                conversion = v4b.ChannelConversion(
                    conversion_type=v4c.CONVERSION_TYPE_RAT,
                    P1=conversion.P1,
                    P2=conversion.P2,
                    P3=conversion.P3,
                    P4=conversion.P4,
                    P5=conversion.P5,
                    P6=conversion.P6,
                )

            elif conversion_type == v3c.CONVERSION_TYPE_FORMULA:
                formula = conversion.formula
                conversion = v4b.ChannelConversion(
                    conversion_type=v4c.CONVERSION_TYPE_ALG, formula=formula
                )

            elif conversion_type == v3c.CONVERSION_TYPE_TAB:
                conversion_ = {}
                conversion_["val_param_nr"] = conversion.ref_param_nr * 2
                for i in range(conversion.ref_param_nr):
                    conversion_[f"raw_{i}"] = conversion[f"raw_{i}"]
                    conversion_[f"phys_{i}"] = conversion[f"phys_{i}"]

                conversion = v4b.ChannelConversion(
                    conversion_type=v4c.CONVERSION_TYPE_TAB, **conversion_
                )

            elif conversion_type == v3c.CONVERSION_TYPE_TABI:
                conversion_ = {}
                conversion_["val_param_nr"] = conversion.ref_param_nr * 2
                for i in range(conversion.ref_param_nr):
                    conversion_[f"raw_{i}"] = conversion[f"raw_{i}"]
                    conversion_[f"phys_{i}"] = conversion[f"phys_{i}"]

                conversion = v4b.ChannelConversion(
                    conversion_type=v4c.CONVERSION_TYPE_TABI, **conversion_
                )

            elif conversion_type == v3c.CONVERSION_TYPE_TABX:

                nr = conversion["ref_param_nr"]
                kargs = {
                    "val_param_nr": nr,
                    "ref_param_nr": nr + 1,
                    "conversion_type": v4c.CONVERSION_TYPE_TABX,
                }
                for i in range(nr):
                    kargs[f"val_{i}"] = conversion[f"param_val_{i}"]
                    kargs[f"text_{i}"] = conversion[f"text_{i}"]

                conversion = v4b.ChannelConversion(**kargs)

            elif conversion_type == v3c.CONVERSION_TYPE_RTABX:

                nr = conversion["ref_param_nr"] - 1
                kargs = {
                    "val_param_nr": nr * 2,
                    "ref_param_nr": nr + 1,
                    "conversion_type": v4c.CONVERSION_TYPE_RTABX,
                    "default_addr": conversion.referenced_blocks["default_addr"],
                }
                for i in range(nr):
                    kargs[f"lower_{i}"] = conversion[f"lower_{i}"]
                    kargs[f"upper_{i}"] = conversion[f"upper_{i}"]
                    kargs[f"text_{i}"] = conversion.referenced_blocks[f"text_{i}"]

                conversion = v4b.ChannelConversion(**kargs)

            conversion.unit = unit

    return conversion class MDF_Common:
    """common methods for MDF objects"""

    def _set_temporary_master(self, master: NDArray[Any] | None) -> None:
        self._master = master

    # @lru_cache(maxsize=1024)
    def _validate_channel_selection(
        self,
        name: str | None = None,
        group: int | None = None,
        index: int | None = None,
    ) -> tuple[int, int]:
        """Gets channel comment.
        Channel can be specified in two ways:

        * using the first positional argument *name*

            * if there are multiple occurrences for this channel then the
            *group* and *index* arguments can be used to select a specific
            group.
            * if there are multiple occurrences for this channel and either the
            *group* or *index* arguments is None then a warning is issued

        * using the group number (keyword argument *group*) and the channel
        number (keyword argument *index*). Use *info* method for group and
        channel numbers


        Parameters
        ----------
        name : string
            name of channel
        group : int
            0-based group index
        index : int
            0-based channel index

        Returns
        -------
        group_index, channel_index : (int, int)
            selected channel's group and channel index

        """

        if name is None:
            if group is None or index is None:
                message = (
                    "Invalid arguments for channel selection: "
                    'must give "name" or, "group" and "index"'
                )
                raise MdfException(message)
            else:
                gp_nr, ch_nr = group, index
                if ch_nr >= 0:
                    try:
                        grp = self.groups[gp_nr]
                    except IndexError:
                        raise MdfException("Group index out of range")

                    try:
                        grp.channels[ch_nr]
                    except IndexError:
                        raise MdfException(
                            f"Channel index out of range: {(name, group, index)}"
                        )
        else:
            if name not in self.channels_db:
                raise MdfException(f'Channel "{name}" not found')
            else:
                if group is None:

                    entries = self.channels_db[name]
                    if len(entries) > 1:
                        if self._raise_on_multiple_occurrences:
                            message = (
                                f'Multiple occurrences for channel "{name}": {entries}. '
                                'Provide both "group" and "index" arguments'
                                " to select another data group"
                            )
                            logger.exception(message)
                            raise MdfException(message)
                        else:
                            message = (
                                f'Multiple occurrences for channel "{name}": {entries}. '
                                "Returning the first occurrence since the MDF object was "
                                "configured to not raise an exception in this case."
                            )
                            logger.warning(message)
                            gp_nr, ch_nr = entries[0]
                    else:
                        gp_nr, ch_nr = entries[0]

                else:
                    if index is not None and index < 0:
                        gp_nr = group
                        ch_nr = index
                    else:
                        if index is None:
                            entries = tuple(
                                (gp_nr, ch_nr)
                                for gp_nr, ch_nr in self.channels_db[name]
                                if gp_nr == group
                            )
                            count = len(entries)

                            if count == 1:
                                gp_nr, ch_nr = entries[0]

                            elif count == 0:
                                message = f'Channel "{name}" not found in group {group}'
                                raise MdfException(message)

                            else:
                                if self._raise_on_multiple_occurrences:
                                    message = (
                                        f'Multiple occurrences for channel "{name}": {entries}. '
                                        'Provide both "group" and "index" arguments'
                                        " to select another data group"
                                    )
                                    logger.exception(message)
                                    raise MdfException(message)
                                else:
                                    message = (
                                        f'Multiple occurrences for channel "{name}": {entries}. '
                                        "Returning the first occurrence since the MDF object was "
                                        "configured to not raise an exception in this case."
                                    )
                                    logger.warning(message)
                                    gp_nr, ch_nr = entries[0]
                        else:
                            if (group, index) in self.channels_db[name]:
                                ch_nr = index
                                gp_nr = group
                            else:
                                message = f'Channel "{name}" not found in group {group} at index {index}'
                                raise MdfException(message)

        return gp_nr, ch_nr def get_global_option(opt):
    return _GLOBAL_OPTIONS[opt] class Source:

    __slots__ = "name", "path", "comment", "source_type", "bus_type"

    SOURCE_OTHER = v4c.SOURCE_OTHER
    SOURCE_ECU = v4c.SOURCE_ECU
    SOURCE_BUS = v4c.SOURCE_BUS
    SOURCE_IO = v4c.SOURCE_IO
    SOURCE_TOOL = v4c.SOURCE_TOOL
    SOURCE_USER = v4c.SOURCE_USER

    BUS_TYPE_NONE = v4c.BUS_TYPE_NONE
    BUS_TYPE_OTHER = v4c.BUS_TYPE_OTHER
    BUS_TYPE_CAN = v4c.BUS_TYPE_CAN
    BUS_TYPE_LIN = v4c.BUS_TYPE_LIN
    BUS_TYPE_MOST = v4c.BUS_TYPE_MOST
    BUS_TYPE_FLEXRAY = v4c.BUS_TYPE_FLEXRAY
    BUS_TYPE_K_LINE = v4c.BUS_TYPE_K_LINE
    BUS_TYPE_ETHERNET = v4c.BUS_TYPE_ETHERNET
    BUS_TYPE_USB = v4c.BUS_TYPE_USB

    def __init__(
        self, name: str, path: str, comment: str, source_type: int, bus_type: int
    ) -> None:
        """Commons reprezentation for source information

        Attributes
        ----------
        name : str
            source name
        path : str
            source path
        comment : str
            source comment
        source_type : int
            source type code
        bus_type : int
            source bus code

        """
        self.name, self.path, self.comment, self.source_type, self.bus_type = (
            name,
            path,
            comment,
            source_type,
            bus_type,
        )

    @classmethod
    @lru_cache(128)
    def from_source(cls, source: SourceType) -> Source:
        if isinstance(source, v3b.ChannelExtension):
            if source.type == v3c.SOURCE_ECU:
                return cls(
                    source.name,
                    source.path,
                    source.comment,
                    cls.SOURCE_OTHER,  # source type other
                    cls.BUS_TYPE_NONE,  # bus type none
                )
            else:
                return cls(
                    source.name,
                    source.path,
                    source.comment,
                    cls.SOURCE_BUS,  # source type bus
                    cls.BUS_TYPE_CAN,  # bus type CAN
                )

        elif isinstance(source, v4b.SourceInformation):
            return cls(
                source.name,
                source.path,
                source.comment,
                source.source_type,
                source.bus_type,
            )
        elif isinstance(source, Source):
            return cls(
                source.name,
                source.path,
                source.comment,
                source.source_type,
                source.bus_type,
            )

    def get_details(self) -> str:
        source_type = v4c.SOURCE_TYPE_TO_STRING[self.source_type]
        bus_type = v4c.BUS_TYPE_TO_STRING[self.bus_type]
        return f"     {source_type} source on bus {bus_type}: name=[{self.name}] path=[{self.path}]" def as_non_byte_sized_signed_int(
    integer_array: NDArray[Any], bit_length: int
) -> NDArray[Any]:
    """
    The MDF spec allows values to be encoded as integers that aren't
    byte-sized. Numpy only knows how to do two's complement on byte-sized
    integers (i.e. int16, int32, int64, etc.), so we have to calculate two's
    complement ourselves in order to handle signed integers with unconventional
    lengths.

    Parameters
    ----------
    integer_array : np.array
        Array of integers to apply two's complement to
    bit_length : int
        Number of bits to sample from the array

    Returns
    -------
    integer_array : np.array
        signed integer array with non-byte-sized two's complement applied

    """

    if integer_array.flags.writeable:
        integer_array &= (1 << bit_length) - 1  # Zero out the unwanted bits
        truncated_integers = integer_array
    else:
        truncated_integers = integer_array & (
            (1 << bit_length) - 1
        )  # Zero out the unwanted bits
    return where(
        truncated_integers
        >> bit_length - 1,  # sign bit as a truth series (True when negative)
        (2**bit_length - truncated_integers)
        * -1,  # when negative, do two's complement
        truncated_integers,  # when positive, return the truncated int
    )
 CHANNEL_COUNT = (1000, 2000, 10000, 20000)
 class ChannelsDB(Dict[str, Tuple[Tuple[int, int], ...]]):
    def __init__(self) -> None:
        super().__init__()

    def add(self, channel_name: str, entry: tuple[int, int]) -> None:
        """add name to channels database and check if it contains a source
        path

        Parameters
        ----------
        channel_name : str
            name that needs to be added to the database
        entry : tuple
            (group index, channel index) pair

        """
        if channel_name:
            if channel_name not in self:
                self[channel_name] = (entry,)
            else:
                self[channel_name] += (entry,)

            if "\\" in channel_name:
                channel_name, _ = channel_name.split("\\", 1)

                if channel_name not in self:
                    self[channel_name] = (entry,)
                elif entry not in self[channel_name]:
                    self[channel_name] += (entry,)
 CONVERT = (10 * 2**20, 20 * 2**20, 30 * 2**20, 40 * 2**20)
 def count_channel_groups(
    stream: ReadableBufferType, include_channels: bool = False, mapped: bool = False
) -> tuple[int, int]:
    """count all channel groups as fast as possible. This is used to provide
    reliable progress information when loading a file using the GUI

    Parameters
    ----------
    stream : file handle
        opened file handle
    include_channels : bool
        also count channels

    Returns
    -------
    count : int
        channel group count

    """

    count = 0
    ch_count = 0

    stream.seek(64)
    blk_id = stream.read(2)
    if blk_id == b"HD":
        version = 3
    else:
        blk_id += stream.read(2)
        if blk_id == b"##HD":
            version = 4
        else:
            raise MdfException(f'"{stream.name}" is not a valid MDF file')

    if version >= 4:
        if mapped:
            dg_addr = UINT64_uf(stream, 88)[0]
            while dg_addr:
                stream.seek(dg_addr + 32)
                cg_addr = UINT64_uf(stream, dg_addr + 32)[0]
                while cg_addr:
                    count += 1
                    if include_channels:
                        ch_addr = UINT64_uf(stream, cg_addr + 32)[0]
                        while ch_addr:
                            ch_count += 1
                            ch_addr = UINT64_uf(stream, ch_addr + 24)[0]
                    cg_addr = UINT64_uf(stream, cg_addr + 24)[0]

                dg_addr = UINT64_uf(stream, dg_addr + 24)[0]
        else:
            stream.seek(88, 0)
            dg_addr = UINT64_u(stream.read(8))[0]
            while dg_addr:
                stream.seek(dg_addr + 32)
                cg_addr = UINT64_u(stream.read(8))[0]
                while cg_addr:
                    count += 1
                    if include_channels:
                        stream.seek(cg_addr + 32)
                        ch_addr = UINT64_u(stream.read(8))[0]
                        while ch_addr:
                            ch_count += 1
                            stream.seek(ch_addr + 24)
                            ch_addr = UINT64_u(stream.read(8))[0]
                    stream.seek(cg_addr + 24)
                    cg_addr = UINT64_u(stream.read(8))[0]

                stream.seek(dg_addr + 24)
                dg_addr = UINT64_u(stream.read(8))[0]

    else:
        stream.seek(68, 0)
        dg_addr = UINT32_u(stream.read(4))[0]
        while dg_addr:
            stream.seek(dg_addr + 8)
            cg_addr = UINT32_u(stream.read(4))[0]
            while cg_addr:
                count += 1
                if include_channels:
                    stream.seek(cg_addr + 8)
                    ch_addr = UINT32_u(stream.read(4))[0]
                    while ch_addr:
                        ch_count += 1
                        stream.seek(ch_addr + 4)
                        ch_addr = UINT32_u(stream.read(4))[0]
                stream.seek(cg_addr + 4)
                cg_addr = UINT32_u(stream.read(4))[0]

            stream.seek(dg_addr + 4)
            dg_addr = UINT32_u(stream.read(4))[0]

    return count, ch_count
 class DataBlockInfo:

    __slots__ = (
        "address",
        "block_type",
        "original_size",
        "compressed_size",
        "param",
        "invalidation_block",
        "block_limit",
    )

    def __init__(
        self,
        address: int,
        block_type: int,
        original_size: int,
        compressed_size: int,
        param: int,
        invalidation_block=None,
        block_limit: int | None = None,
    ) -> None:
        self.address = address
        self.block_type = block_type
        self.original_size = original_size
        self.compressed_size = compressed_size
        self.param = param
        self.invalidation_block = invalidation_block
        self.block_limit = block_limit

    def __repr__(self) -> str:
        return (
            f"DataBlockInfo(address=0x{self.address:X}, "
            f"block_type={self.block_type}, "
            f"original_size={self.original_size}, "
            f"compressed_size={self.compressed_size}, "
            f"param={self.param}, "
            f"invalidation_block={self.invalidation_block}, "
            f"block_limit={self.block_limit})"
        )
 @lru_cache(maxsize=1024)
def fmt_to_datatype_v3(
    fmt: dtype[Any], shape: tuple[int, ...], array: bool = False
) -> tuple[int, int]:
    """convert numpy dtype format string to mdf versions 2 and 3
    channel data type and size

    Parameters
    ----------
    fmt : numpy.dtype
        numpy data type
    shape : tuple
        numpy array shape
    array : bool
        disambiguate between bytearray and channel array

    Returns
    -------
    data_type, size : int, int
        integer data type as defined by ASAM MDF and bit size

    """
    byteorder = fmt.byteorder
    if byteorder in "=|":
        byteorder = "<" if sys.byteorder == "little" else ">"
    size = fmt.itemsize * 8
    kind = fmt.kind

    if not array and shape[1:] and fmt.itemsize == 1 and kind == "u":
        data_type = v3c.DATA_TYPE_BYTEARRAY
        for dim in shape[1:]:
            size *= dim
    else:
        if kind == "u":
            if byteorder in "<":
                data_type = v3c.DATA_TYPE_UNSIGNED_INTEL
            else:
                data_type = v3c.DATA_TYPE_UNSIGNED_MOTOROLA
        elif kind == "i":
            if byteorder in "<":
                data_type = v3c.DATA_TYPE_SIGNED_INTEL
            else:
                data_type = v3c.DATA_TYPE_SIGNED_MOTOROLA
        elif kind == "f":
            if byteorder in "<":
                if size == 32:
                    data_type = v3c.DATA_TYPE_FLOAT
                else:
                    data_type = v3c.DATA_TYPE_DOUBLE
            else:
                if size == 32:
                    data_type = v3c.DATA_TYPE_FLOAT_MOTOROLA
                else:
                    data_type = v3c.DATA_TYPE_DOUBLE_MOTOROLA
        elif kind in "SV":
            data_type = v3c.DATA_TYPE_STRING
        elif kind == "b":
            data_type = v3c.DATA_TYPE_UNSIGNED_INTEL
            size = 1
        else:
            message = f"Unknown type: dtype={fmt}, shape={shape}"
            logger.exception(message)
            raise MdfException(message)

    return data_type, size
 @lru_cache(maxsize=1024)
def get_fmt_v3(
    data_type: int, size: int, byte_order: int = v3c.BYTE_ORDER_INTEL
) -> str:
    """convert mdf versions 2 and 3 channel data type to numpy dtype format
    string

    Parameters
    ----------
    data_type : int
        mdf channel data type
    size : int
        data bit size
    Returns
    -------
    fmt : str
        numpy compatible data type format string

    """
    if data_type in (v3c.DATA_TYPE_STRING, v3c.DATA_TYPE_BYTEARRAY):
        size = size // 8
        if data_type == v3c.DATA_TYPE_STRING:
            fmt = f"S{size}"
        else:
            fmt = f"({size},)u1"
    else:
        if size > 64 and data_type in (
            v3c.DATA_TYPE_UNSIGNED_INTEL,
            v3c.DATA_TYPE_UNSIGNED,
            v3c.DATA_TYPE_UNSIGNED_MOTOROLA,
        ):
            fmt = f"({size // 8},)u1"
        else:
            if size <= 8:
                size = 1
            elif size <= 16:
                size = 2
            elif size <= 32:
                size = 4
            elif size <= 64:
                size = 8
            else:
                size = size // 8

            if data_type == v3c.DATA_TYPE_UNSIGNED_INTEL:
                fmt = f"<u{size}"

            elif data_type == v3c.DATA_TYPE_UNSIGNED:
                if byte_order == v3c.BYTE_ORDER_INTEL:
                    fmt = f"<u{size}"
                else:
                    fmt = f">u{size}"

            elif data_type == v3c.DATA_TYPE_UNSIGNED_MOTOROLA:
                fmt = f">u{size}"

            elif data_type == v3c.DATA_TYPE_SIGNED_INTEL:
                fmt = f"<i{size}"

            elif data_type == v3c.DATA_TYPE_SIGNED:
                if byte_order == v3c.BYTE_ORDER_INTEL:
                    fmt = f"<i{size}"
                else:
                    fmt = f">i{size}"

            elif data_type == v3c.DATA_TYPE_SIGNED_MOTOROLA:
                fmt = f">i{size}"

            elif data_type in (v3c.DATA_TYPE_FLOAT_INTEL, v3c.DATA_TYPE_DOUBLE_INTEL):
                fmt = f"<f{size}"

            elif data_type in (
                v3c.DATA_TYPE_FLOAT_MOTOROLA,
                v3c.DATA_TYPE_DOUBLE_MOTOROLA,
            ):
                fmt = f">f{size}"

            elif data_type in (v3c.DATA_TYPE_FLOAT, v3c.DATA_TYPE_DOUBLE):
                if byte_order == v3c.BYTE_ORDER_INTEL:
                    fmt = f"<f{size}"
                else:
                    fmt = f">f{size}"

    return fmt
 @overload
def get_text_v3(
    address: int,
    stream: ReadableBufferType,
    mapped: bool = ...,
    decode: Literal[True] = ...,
) -> str:
    ...
 class Group:

    __slots__ = (
        "channels",
        "channel_dependencies",
        "signal_data",
        "channel_group",
        "record_size",
        "sorted",
        "data_group",
        "data_location",
        "data_blocks",
        "data_blocks_info_generator",
        "record_size",
        "record",
        "signal_types",
        "trigger",
        "string_dtypes",
        "single_channel_dtype",
        "uses_ld",
        "read_split_count",
    )

    def __init__(self, data_group: DataGroupType) -> None:
        self.data_group = data_group
        self.channels = []
        self.channel_dependencies = []
        self.signal_data = []
        self.record = None
        self.trigger = None
        self.string_dtypes = None
        self.data_blocks = []
        self.single_channel_dtype = None
        self.uses_ld = False
        self.read_split_count = 0
        self.data_blocks_info_generator = iter(EMPTY_TUPLE)

    def __getitem__(self, item: str) -> Any:
        return self.__getattribute__(item)

    def __setitem__(self, item: str, value: Any) -> None:
        self.__setattr__(item, value)

    def set_blocks_info(self, info: list[DataBlockInfo]) -> None:
        self.data_blocks = info

    def __contains__(self, item: str) -> bool:
        return hasattr(self, item)

    def clear(self) -> None:
        self.data_blocks.clear()
        self.channels.clear()
        self.channel_dependencies.clear()
        self.signal_data.clear()
        self.data_blocks_info_generator = None

    def get_data_blocks(self) -> Iterator[DataBlockInfo]:
        for blk in self.data_blocks:
            yield blk

        while True:
            try:
                info = next(self.data_blocks_info_generator)
                self.data_blocks.append(info)
                yield info
            except StopIteration:
                break

    def get_signal_data_blocks(self, index: int) -> Iterator[SignalDataBlockInfo]:
        signal_data = self.signal_data[index]
        if signal_data is not None:
            signal_data, signal_generator = signal_data
            for blk in signal_data:
                yield blk

            while True:
                try:
                    info = next(signal_generator)
                    signal_data.append(info)
                    yield info
                except StopIteration:
                    break
 def is_file_like(obj: object) -> bool:
    """
    Check if the object is a file-like object.

    For objects to be considered file-like, they must
    be an iterator AND have a 'read' and 'seek' method
    as an attribute.

    Note: file-like objects must be iterable, but
    iterable objects need not be file-like.

    Parameters
    ----------
    obj : The object to check.

    Returns
    -------
    is_file_like : bool
        Whether `obj` has file-like properties.

    Examples
    --------
    >>> buffer(StringIO("data"))
    >>> is_file_like(buffer)
    True
    >>> is_file_like([1, 2, 3])
    False
    """
    if not (hasattr(obj, "read") and hasattr(obj, "seek")):
        return False

    if not hasattr(obj, "__iter__"):
        return False

    return True
 class MdfException(Exception):
    """MDF Exception class"""

    pass
 class UniqueDB(object):
    def __init__(self) -> None:
        self._db = {}

    def get_unique_name(self, name: str) -> str:
        """returns an available unique name

        Parameters
        ----------
        name : str
            name to be made unique

        Returns
        -------
        unique_name : str
            new unique name

        """

        if name not in self._db:
            self._db[name] = 0
            return name
        else:
            index = self._db[name]
            self._db[name] = index + 1
            return f"{name}_{index}"
 def validate_version_argument(version: str, hint: int = 4) -> str:
    """validate the version argument against the supported MDF versions. The
    default version used depends on the hint MDF major revision

    Parameters
    ----------
    version : str
        requested MDF version
    hint : int
        MDF revision hint

    Returns
    -------
    valid_version : str
        valid version

    """
    if version not in SUPPORTED_VERSIONS:
        if hint == 2:
            valid_version = "2.14"
        elif hint == 3:
            valid_version = "3.30"
        else:
            valid_version = "4.10"
        message = (
            'Unknown mdf version "{}".'
            " The available versions are {};"
            ' automatically using version "{}"'
        )
        message = message.format(version, SUPPORTED_VERSIONS, valid_version)
        logger.warning(message)
    else:
        valid_version = version
    return valid_version
 class VirtualChannelGroup:
    """starting with MDF v4.20 it is possible to use remote masters and column
    oriented storage. This means we now have virtual channel groups that can
    span over multiple regular channel groups. This class facilitates the
    handling of this virtual groups"""

    __slots__ = (
        "groups",
        "record_size",
        "cycles_nr",
    )

    def __init__(self) -> None:
        self.groups = []
        self.record_size = 0
        self.cycles_nr = 0

    def __repr__(self) -> None:
        return f"VirtualChannelGroup(groups={self.groups}, records_size={self.record_size}, cycles_nr={self.cycles_nr})"
 class Channel:
    """CNBLOCK class

    If the `load_metadata` keyword argument is not provided or is False,
    then the conversion, source and display name information is not processed.

    CNBLOCK fields

    * ``id`` - bytes : block ID; always b'CN'
    * ``block_len`` - int : block bytes size
    * ``next_ch_addr`` - int : next CNBLOCK address
    * ``conversion_addr`` - int : address of channel conversion block
    * ``source_addr`` - int : address of channel source block
    * ``component_addr`` - int : address of dependency block (CDBLOCK) of this
      channel
    * ``comment_addr`` - int : address of TXBLOCK that contains the
      channel comment
    * ``channel_type`` - int : integer code for channel type
    * ``short_name`` - bytes : short signal name
    * ``description`` - bytes : signal description
    * ``start_offset`` - int : start offset in bits to determine the first bit
      of the signal in the data record
    * ``bit_count`` - int : channel bit count
    * ``data_type`` - int : integer code for channel data type
    * ``range_flag`` - int : value range valid flag
    * ``min_raw_value`` - float : min raw value of all samples
    * ``max_raw_value`` - float : max raw value of all samples
    * ``sampling_rate`` - float : sampling rate in *'s'* for a virtual time
      channel
    * ``long_name_addr`` - int : address of TXBLOCK that contains the channel's
      name
    * ``display_name_addr`` - int : address of TXBLOCK that contains the
      channel's display name
    * ``aditional_byte_offset`` - int : additional Byte offset of the channel
      in the data record

    Other attributes

    * ``address`` - int : block address inside mdf file
    * ``comment`` - str : channel comment
    * ``conversion`` - ChannelConversion : channel conversion; *None* if the channel has
      no conversion
    * ``display_names`` - dict : channel display names

        ..versionchanged:: 7.0.0

            changed from str to dict

    * ``name`` - str : full channel name
    * ``source`` - SourceInformation : channel source information; *None* if the channel
      has no source information

    Parameters
    ----------
    address : int
        block address; to be used for objects created from file
    stream : handle
        file handle; to be used for objects created from file
    load_metadata : bool
        option to load conversion, source and display_names; default *True*
    for dynamically created objects :
        see the key-value pairs

    Examples
    --------
    >>> with open('test.mdf', 'rb') as mdf:
    ...     ch1 = Channel(stream=mdf, address=0xBA52)
    >>> ch2 = Channel()
    >>> ch1.name
    'VehicleSpeed'
    >>> ch1['id']
    b'CN'

    """

    __slots__ = (
        "name",
        "display_names",
        "comment",
        "conversion",
        "source",
        "address",
        "id",
        "block_len",
        "next_ch_addr",
        "conversion_addr",
        "source_addr",
        "component_addr",
        "comment_addr",
        "channel_type",
        "short_name",
        "description",
        "start_offset",
        "bit_count",
        "data_type",
        "range_flag",
        "min_raw_value",
        "max_raw_value",
        "sampling_rate",
        "long_name_addr",
        "display_name_addr",
        "additional_byte_offset",
        "dtype_fmt",
    )

    def __init__(self, **kwargs) -> None:
        super().__init__()

        self.name = self.comment = ""
        self.display_names = {}
        self.conversion = self.source = None
        self.dtype_fmt = None

        try:
            stream = kwargs["stream"]
            self.address = address = kwargs["address"]
            mapped = kwargs.get("mapped", False)

            if mapped:
                (size,) = UINT16_uf(stream, address + 2)

                if size == v23c.CN_DISPLAYNAME_BLOCK_SIZE:
                    (
                        self.id,
                        self.block_len,
                        self.next_ch_addr,
                        self.conversion_addr,
                        self.source_addr,
                        self.component_addr,
                        self.comment_addr,
                        self.channel_type,
                        self.short_name,
                        self.description,
                        self.start_offset,
                        self.bit_count,
                        self.data_type,
                        self.range_flag,
                        self.min_raw_value,
                        self.max_raw_value,
                        self.sampling_rate,
                        self.long_name_addr,
                        self.display_name_addr,
                        self.additional_byte_offset,
                    ) = CHANNEL_DISPLAYNAME_uf(stream, address)

                    parsed_strings = kwargs["parsed_strings"]

                    if parsed_strings is None:

                        addr = self.long_name_addr
                        if addr:
                            self.name = get_text_v3(
                                address=addr, stream=stream, mapped=mapped
                            )
                        else:
                            self.name = self.short_name.decode("latin-1").strip(
                                " \t\n\r\0"
                            )

                        addr = self.display_name_addr
                        if addr:
                            self.display_names = {
                                get_text_v3(
                                    address=addr, stream=stream, mapped=mapped
                                ): "display_name",
                            }

                    else:
                        self.name, self.display_names = parsed_strings

                elif size == v23c.CN_LONGNAME_BLOCK_SIZE:
                    (
                        self.id,
                        self.block_len,
                        self.next_ch_addr,
                        self.conversion_addr,
                        self.source_addr,
                        self.component_addr,
                        self.comment_addr,
                        self.channel_type,
                        self.short_name,
                        self.description,
                        self.start_offset,
                        self.bit_count,
                        self.data_type,
                        self.range_flag,
                        self.min_raw_value,
                        self.max_raw_value,
                        self.sampling_rate,
                        self.long_name_addr,
                    ) = CHANNEL_LONGNAME_uf(stream, address)

                    parsed_strings = kwargs["parsed_strings"]

                    if parsed_strings is None:

                        addr = self.long_name_addr
                        if addr:
                            self.name = get_text_v3(
                                address=addr, stream=stream, mapped=mapped
                            )
                        else:
                            self.name = self.short_name.decode("latin-1").strip(
                                " \t\n\r\0"
                            )

                    else:
                        self.name, self.display_names = parsed_strings

                else:
                    (
                        self.id,
                        self.block_len,
                        self.next_ch_addr,
                        self.conversion_addr,
                        self.source_addr,
                        self.component_addr,
                        self.comment_addr,
                        self.channel_type,
                        self.short_name,
                        self.description,
                        self.start_offset,
                        self.bit_count,
                        self.data_type,
                        self.range_flag,
                        self.min_raw_value,
                        self.max_raw_value,
                        self.sampling_rate,
                    ) = CHANNEL_SHORT_uf(stream, address)

                    self.name = self.short_name.decode("latin-1").strip(" \t\n\r\0")

                cc_map = kwargs.get("cc_map", {})
                si_map = kwargs.get("si_map", {})

                address = self.conversion_addr
                if address:
                    try:
                        if address in cc_map:
                            conv = cc_map[address]
                        else:
                            (size,) = UINT16_uf(stream, address + 2)
                            raw_bytes = stream[address : address + size]

                            if raw_bytes in cc_map:
                                conv = cc_map[raw_bytes]
                            else:
                                conv = ChannelConversion(
                                    raw_bytes=raw_bytes,
                                    stream=stream,
                                    address=address,
                                    mapped=mapped,
                                )
                                cc_map[raw_bytes] = cc_map[address] = conv
                    except:
                        logger.warning(
                            f"Channel conversion parsing error: {format_exc()}. The error is ignored and the channel conversion is None"
                        )
                        conv = None
                    self.conversion = conv

                address = self.source_addr
                if address:
                    try:
                        if address in si_map:
                            source = si_map[address]
                        else:
                            raw_bytes = stream[address : address + v23c.CE_BLOCK_SIZE]

                            if raw_bytes in si_map:
                                source = si_map[raw_bytes]
                            else:
                                source = ChannelExtension(
                                    raw_bytes=raw_bytes,
                                    stream=stream,
                                    address=address,
                                    mapped=mapped,
                                )
                                si_map[raw_bytes] = si_map[address] = source
                    except:
                        logger.warning(
                            f"Channel source parsing error: {format_exc()}. The error is ignored and the channel source is None"
                        )
                        source = None

                    self.source = source

                self.comment = get_text_v3(
                    address=self.comment_addr, stream=stream, mapped=mapped
                )
            else:
                stream.seek(address + 2)
                (size,) = UINT16_u(stream.read(2))
                stream.seek(address)
                block = stream.read(size)

                if size == v23c.CN_DISPLAYNAME_BLOCK_SIZE:
                    (
                        self.id,
                        self.block_len,
                        self.next_ch_addr,
                        self.conversion_addr,
                        self.source_addr,
                        self.component_addr,
                        self.comment_addr,
                        self.channel_type,
                        self.short_name,
                        self.description,
                        self.start_offset,
                        self.bit_count,
                        self.data_type,
                        self.range_flag,
                        self.min_raw_value,
                        self.max_raw_value,
                        self.sampling_rate,
                        self.long_name_addr,
                        self.display_name_addr,
                        self.additional_byte_offset,
                    ) = CHANNEL_DISPLAYNAME_u(block)

                    parsed_strings = kwargs["parsed_strings"]

                    if parsed_strings is None:

                        addr = self.long_name_addr
                        if addr:
                            self.name = get_text_v3(
                                address=addr, stream=stream, mapped=mapped
                            )
                        else:
                            self.name = self.short_name.decode("latin-1").strip(
                                " \t\n\r\0"
                            )

                        addr = self.display_name_addr
                        if addr:
                            self.display_names = {
                                get_text_v3(
                                    address=addr, stream=stream, mapped=mapped
                                ): "display_name",
                            }

                    else:
                        self.name, self.display_names = parsed_strings

                elif size == v23c.CN_LONGNAME_BLOCK_SIZE:
                    (
                        self.id,
                        self.block_len,
                        self.next_ch_addr,
                        self.conversion_addr,
                        self.source_addr,
                        self.component_addr,
                        self.comment_addr,
                        self.channel_type,
                        self.short_name,
                        self.description,
                        self.start_offset,
                        self.bit_count,
                        self.data_type,
                        self.range_flag,
                        self.min_raw_value,
                        self.max_raw_value,
                        self.sampling_rate,
                        self.long_name_addr,
                    ) = CHANNEL_LONGNAME_u(block)

                    parsed_strings = kwargs["parsed_strings"]

                    if parsed_strings is None:

                        addr = self.long_name_addr
                        if addr:
                            self.name = get_text_v3(
                                address=addr, stream=stream, mapped=mapped
                            )
                        else:
                            self.name = self.short_name.decode("latin-1").strip(
                                " \t\n\r\0"
                            )

                    else:
                        self.name, self.display_names = parsed_strings

                else:
                    (
                        self.id,
                        self.block_len,
                        self.next_ch_addr,
                        self.conversion_addr,
                        self.source_addr,
                        self.component_addr,
                        self.comment_addr,
                        self.channel_type,
                        self.short_name,
                        self.description,
                        self.start_offset,
                        self.bit_count,
                        self.data_type,
                        self.range_flag,
                        self.min_raw_value,
                        self.max_raw_value,
                        self.sampling_rate,
                    ) = CHANNEL_SHORT_u(block)

                    self.name = self.short_name.decode("latin-1").strip(" \t\n\r\0")

                cc_map = kwargs.get("cc_map", {})
                si_map = kwargs.get("si_map", {})

                address = self.conversion_addr
                if address:
                    try:
                        if address in cc_map:
                            conv = cc_map[address]
                        else:
                            stream.seek(address + 2)
                            (size,) = UINT16_u(stream.read(2))
                            stream.seek(address)
                            raw_bytes = stream.read(size)

                            if raw_bytes in cc_map:
                                conv = cc_map[raw_bytes]
                            else:
                                conv = ChannelConversion(
                                    raw_bytes=raw_bytes,
                                    stream=stream,
                                    address=address,
                                    mapped=mapped,
                                )
                                cc_map[raw_bytes] = cc_map[address] = conv
                    except:
                        logger.warning(
                            f"Channel conversion parsing error: {format_exc()}. The error is ignored and the channel conversion is None"
                        )
                        conv = None

                    self.conversion = conv

                address = self.source_addr
                if address:
                    try:
                        if address in si_map:
                            source = si_map[address]
                        else:
                            stream.seek(address)
                            raw_bytes = stream.read(v23c.CE_BLOCK_SIZE)

                            if raw_bytes in si_map:
                                source = si_map[raw_bytes]
                            else:
                                source = ChannelExtension(
                                    raw_bytes=raw_bytes,
                                    stream=stream,
                                    address=address,
                                    mapped=mapped,
                                )
                                si_map[raw_bytes] = si_map[address] = source
                    except:
                        logger.warning(
                            f"Channel source parsing error: {format_exc()}. The error is ignored and the channel source is None"
                        )
                        source = None
                    self.source = source

                self.comment = get_text_v3(
                    address=self.comment_addr, stream=stream, mapped=mapped
                )

            if self.id != b"CN":
                message = f'Expected "CN" block @{hex(address)} but found "{self.id}"'
                logger.exception(message)
                raise MdfException(message)

        except KeyError:

            self.address = 0
            self.id = b"CN"
            self.block_len = kwargs.get("block_len", v23c.CN_DISPLAYNAME_BLOCK_SIZE)
            self.next_ch_addr = kwargs.get("next_ch_addr", 0)
            self.conversion_addr = kwargs.get("conversion_addr", 0)
            self.source_addr = kwargs.get("source_addr", 0)
            self.component_addr = kwargs.get("component_addr", 0)
            self.comment_addr = kwargs.get("comment_addr", 0)
            self.channel_type = kwargs.get("channel_type", 0)
            self.short_name = kwargs.get("short_name", (b"\0" * 32))
            self.description = kwargs.get("description", (b"\0" * 128))
            self.start_offset = kwargs.get("start_offset", 0)
            self.bit_count = kwargs.get("bit_count", 8)
            self.data_type = kwargs.get("data_type", 0)
            self.range_flag = kwargs.get("range_flag", 1)
            self.min_raw_value = kwargs.get("min_raw_value", 0)
            self.max_raw_value = kwargs.get("max_raw_value", 0)
            self.sampling_rate = kwargs.get("sampling_rate", 0)
            if self.block_len >= v23c.CN_LONGNAME_BLOCK_SIZE:
                self.long_name_addr = kwargs.get("long_name_addr", 0)
            if self.block_len >= v23c.CN_DISPLAYNAME_BLOCK_SIZE:
                self.display_name_addr = kwargs.get("display_name_addr", 0)
                self.additional_byte_offset = kwargs.get("additional_byte_offset", 0)

        if self.name in self.display_names:
            del self.display_names[self.name]

    def to_blocks(
        self,
        address: int,
        blocks: list[Any],
        defined_texts: dict[str, int],
        cc_map: dict[bytes, int],
        si_map: dict[bytes, int],
    ) -> int:
        key = "long_name_addr"
        text = self.name
        if self.block_len >= v23c.CN_LONGNAME_BLOCK_SIZE:
            if len(text) > 31:
                if text in defined_texts:
                    self[key] = defined_texts[text]
                else:
                    tx_block = TextBlock(text=text)
                    self[key] = address
                    defined_texts[text] = address
                    tx_block.address = address
                    address += tx_block.block_len
                    blocks.append(tx_block)
            else:
                self[key] = 0

        self.short_name = text.encode("latin-1", "backslashreplace")[:31]

        key = "display_name_addr"
        text = list(self.display_names)[0] if self.display_names else ""
        if self.block_len >= v23c.CN_DISPLAYNAME_BLOCK_SIZE:
            if text:
                if text in defined_texts:
                    self[key] = defined_texts[text]
                else:
                    tx_block = TextBlock(text=text)
                    self[key] = address
                    defined_texts[text] = address
                    tx_block.address = address
                    address += tx_block.block_len
                    blocks.append(tx_block)
            else:
                self[key] = 0

        key = "comment_addr"
        text = self.comment
        if text:
            if len(text) < 128:
                self.description = text.encode("latin-1", "backslashreplace")[:127]
                self[key] = 0
            else:
                if text in defined_texts:
                    self[key] = defined_texts[text]
                else:
                    tx_block = TextBlock(text=text)
                    self[key] = address
                    defined_texts[text] = address
                    tx_block.address = address
                    address += tx_block.block_len
                    blocks.append(tx_block)
                self.description = b"\0"
        else:
            self[key] = 0

        conversion = self.conversion
        if conversion:
            address = conversion.to_blocks(address, blocks, defined_texts, cc_map)
            self.conversion_addr = conversion.address
        else:
            self.conversion_addr = 0

        source = self.source
        if source:
            address = source.to_blocks(address, blocks, defined_texts, si_map)
            self.source_addr = source.address
        else:
            self.source_addr = 0

        blocks.append(self)
        self.address = address
        address += self.block_len

        return address

    def metadata(self) -> str:
        max_len = max(len(key) for key in self)
        template = f"{{: <{max_len}}}: {{}}"

        metadata = []
        lines = f"""
name: {self.name}
display names: {self.display_names}
address: {hex(self.address)}
comment: {self.comment}

""".split(
            "\n"
        )

        keys = (
            "id",
            "block_len",
            "next_ch_addr",
            "conversion_addr",
            "source_addr",
            "component_addr",
            "comment_addr",
            "channel_type",
            "short_name",
            "description",
            "start_offset",
            "bit_count",
            "data_type",
            "range_flag",
            "min_raw_value",
            "max_raw_value",
            "sampling_rate",
            "long_name_addr",
            "display_name_addr",
            "additional_byte_offset",
        )

        for key in keys:
            if not hasattr(self, key):
                continue
            val = getattr(self, key)
            if key.endswith("addr") or key.startswith("text_"):
                lines.append(template.format(key, hex(val)))
            elif isinstance(val, float):
                lines.append(template.format(key, round(val, 6)))
            else:
                if isinstance(val, bytes):
                    lines.append(template.format(key, val.strip(b"\0")))
                else:
                    lines.append(template.format(key, val))

            if key == "data_type":
                lines[-1] += f" = {v23c.DATA_TYPE_TO_STRING[self.data_type]}"
            elif key == "channel_type":
                lines[-1] += f" = {v23c.CHANNEL_TYPE_TO_STRING[self.channel_type]}"

        for line in lines:
            if not line:
                metadata.append(line)
            else:
                for wrapped_line in wrap(line, width=120):
                    metadata.append(wrapped_line)

        return "\n".join(metadata)

    def __bytes__(self) -> bytes:

        block_len = self.block_len
        if block_len == v23c.CN_DISPLAYNAME_BLOCK_SIZE:
            return v23c.CHANNEL_DISPLAYNAME_p(
                self.id,
                self.block_len,
                self.next_ch_addr,
                self.conversion_addr,
                self.source_addr,
                self.component_addr,
                self.comment_addr,
                self.channel_type,
                self.short_name,
                self.description,
                self.start_offset,
                self.bit_count,
                self.data_type,
                self.range_flag,
                self.min_raw_value,
                self.max_raw_value,
                self.sampling_rate,
                self.long_name_addr,
                self.display_name_addr,
                self.additional_byte_offset,
            )
        elif block_len == v23c.CN_LONGNAME_BLOCK_SIZE:
            return v23c.CHANNEL_LONGNAME_p(
                self.id,
                self.block_len,
                self.next_ch_addr,
                self.conversion_addr,
                self.source_addr,
                self.component_addr,
                self.comment_addr,
                self.channel_type,
                self.short_name,
                self.description,
                self.start_offset,
                self.bit_count,
                self.data_type,
                self.range_flag,
                self.min_raw_value,
                self.max_raw_value,
                self.sampling_rate,
                self.long_name_addr,
            )
        else:
            return v23c.CHANNEL_SHORT_p(
                self.id,
                self.block_len,
                self.next_ch_addr,
                self.conversion_addr,
                self.source_addr,
                self.component_addr,
                self.comment_addr,
                self.channel_type,
                self.short_name,
                self.description,
                self.start_offset,
                self.bit_count,
                self.data_type,
                self.range_flag,
                self.min_raw_value,
                self.max_raw_value,
                self.sampling_rate,
            )

    def __getitem__(self, item: str) -> Any:
        return self.__getattribute__(item)

    def __setitem__(self, item: str, value: Any) -> None:
        self.__setattr__(item, value)

    def __contains__(self, item: str) -> bool:
        return hasattr(self, item)

    def __iter__(self) -> Iterator[Any]:
        for attr in dir(self):
            if attr[:2] + attr[-2:] == "____":
                continue
            try:
                if callable(getattr(self, attr)):
                    continue
                yield attr
            except AttributeError:
                continue

    def __lt__(self, other: Channel) -> bool:
        self_start = self.start_offset
        other_start = other.start_offset
        try:
            self_additional_offset = self.additional_byte_offset
            if self_additional_offset:
                self_start += 8 * self_additional_offset
            other_additional_offset = other.additional_byte_offset
            if other_additional_offset:
                other_start += 8 * other_additional_offset
        except AttributeError:
            pass

        if self_start < other_start:
            result = 1
        elif self_start == other_start:
            if self.bit_count >= other.bit_count:
                result = 1
            else:
                result = 0
        else:
            result = 0
        return result

    def __repr__(self) -> str:
        fields = []
        for attr in dir(self):
            if attr[:2] + attr[-2:] == "____":
                continue
            try:
                if callable(getattr(self, attr)):
                    continue
                fields.append(f"{attr}:{getattr(self, attr)}")
            except AttributeError:
                continue
        return f"Channel (name: {self.name}, display names: {self.display_names}, comment: {self.comment}, address: {hex(self.address)}, fields: {fields})" class ChannelConversion(_ChannelConversionBase):
    """CCBLOCK class

    *ChannelConversion* has the following common fields

    * ``id`` - bytes : block ID; always b'CC'
    * ``block_len`` - int : block bytes size
    * ``range_flag`` - int : value range valid flag
    * ``min_phy_value`` - float : min raw value of all samples
    * ``max_phy_value`` - float : max raw value of all samples
    * ``unit`` - bytes : physical unit
    * ``conversion_type`` - int : integer code for conversion type
    * ``ref_param_nr`` - int : number of referenced parameters

    *ChannelConversion* has the following specific fields

    * linear conversion

        * ``a`` - float : factor
        * ``b`` - float : offset
        * ``CANapeHiddenExtra`` - bytes : sometimes CANape appends extra
          information; not compliant with MDF specs

    * algebraic conversion

        * ``formula`` - bytes : equation as string

    * polynomial or rational conversion

        * ``P1`` to ``P6`` - float : parameters

    * exponential or logarithmic conversion

        * ``P1`` to ``P7`` - float : parameters

    * tabular with or without interpolation (grouped by index)

        * ``raw_<N>`` - int : N-th raw value (X axis)
        * ``phys_<N>`` - float : N-th physical value (Y axis)

    * text table conversion

        * ``param_val_<N>`` - int : N-th raw value (X axis)
        * ``text_<N>`` - N-th text physical value (Y axis)

    * text range table conversion

        * ``default_lower`` - float : default lower raw value
        * ``default_upper`` - float : default upper raw value
        * ``default_addr`` - int : address of default text physical value
        * ``lower_<N>`` - float : N-th lower raw value
        * ``upper_<N>`` - float : N-th upper raw value
        * ``text_<N>`` - int : address of N-th text physical value

    Other attributes

    * ``address`` - int : block address inside mdf file
    * ``formula`` - str : formula string in case of algebraic conversion
    * ``referenced_blocks`` - list : list of CCBLOCK/TXBLOCK referenced by the conversion
    * ``unit`` - str : physical unit

    Parameters
    ----------
    address : int
        block address inside mdf file
    raw_bytes : bytes
        complete block read from disk
    stream : file handle
        mdf file handle
    for dynamically created objects :
        see the key-value pairs

    Examples
    --------
    >>> with open('test.mdf', 'rb') as mdf:
    ...     cc1 = ChannelConversion(stream=mdf, address=0xBA52)
    >>> cc2 = ChannelConversion(conversion_type=0)
    >>> cc1['b'], cc1['a']
    0, 100.0

    """

    def __init__(self, **kwargs) -> None:
        super().__init__()

        self.unit = self.formula = ""

        self.referenced_blocks = {}

        if "raw_bytes" in kwargs or "stream" in kwargs:
            mapped = kwargs.get("mapped", False)
            self.address = address = kwargs.get("address", 0)
            try:
                block = kwargs["raw_bytes"]
                (self.id, self.block_len) = COMMON_uf(block)
                size = self.block_len
                block_size = len(block)
                block = block[4:]
                stream = kwargs["stream"]

            except KeyError:
                stream = kwargs["stream"]
                if mapped:
                    (self.id, self.block_len) = COMMON_uf(stream, address)
                    block_size = size = self.block_len
                    block = stream[address + 4 : address + size]
                else:
                    stream.seek(address)
                    block = stream.read(4)
                    (self.id, self.block_len) = COMMON_u(block)

                    block_size = size = self.block_len
                    block = stream.read(size - 4)

            (
                self.range_flag,
                self.min_phy_value,
                self.max_phy_value,
                self.unit_field,
                self.conversion_type,
                self.ref_param_nr,
            ) = CONVERSION_COMMON_SHORT_uf(block)

            self.unit = self.unit_field.decode("latin-1").strip(" \t\r\n\0")

            conv_type = self.conversion_type

            if conv_type == v23c.CONVERSION_TYPE_LINEAR:
                (self.b, self.a) = unpack_from("<2d", block, v23c.CC_COMMON_SHORT_SIZE)
                if not size == v23c.CC_LIN_BLOCK_SIZE:
                    self.CANapeHiddenExtra = block[v23c.CC_LIN_BLOCK_SIZE - 4 :]
                    size = len(self.CANapeHiddenExtra)
                    nr = size // 40
                    values = unpack_from(
                        "<" + "d32s" * nr, block, v23c.CC_COMMON_SHORT_SIZE
                    )

                    for i in range(nr):
                        (self[f"param_val_{i}"], self[f"text_{i}"]) = (
                            values[i * 2],
                            values[2 * i + 1],
                        )

            elif conv_type == v23c.CONVERSION_TYPE_NONE:
                pass

            elif conv_type == v23c.CONVERSION_TYPE_FORMULA:
                self.formula_field = block[v23c.CC_COMMON_SHORT_SIZE :]
                self.formula = (
                    self.formula_field.decode("latin-1")
                    .strip(" \t\r\n\0")
                    .replace("x", "X")
                )
                if "X1" not in self.formula:
                    self.formula = self.formula.replace("X", "X1")

            elif conv_type in (v23c.CONVERSION_TYPE_TABI, v23c.CONVERSION_TYPE_TAB):

                nr = self.ref_param_nr

                size = v23c.CC_COMMON_BLOCK_SIZE + nr * 16

                if block_size == v23c.MAX_UINT16:
                    stream.seek(address)
                    raw_bytes = stream.read(size)
                    conversion = ChannelConversion(
                        raw_bytes=raw_bytes, stream=stream, address=address
                    )
                    conversion.block_len = size

                    self.update(conversion)
                    self.referenced_blocks = conversion.referenced_blocks

                else:
                    values = unpack_from(f"<{2*nr}d", block, v23c.CC_COMMON_SHORT_SIZE)
                    for i in range(nr):
                        (self[f"raw_{i}"], self[f"phys_{i}"]) = (
                            values[i * 2],
                            values[2 * i + 1],
                        )

            elif conv_type in (v23c.CONVERSION_TYPE_POLY, v23c.CONVERSION_TYPE_RAT):
                (self.P1, self.P2, self.P3, self.P4, self.P5, self.P6) = unpack_from(
                    "<6d", block, v23c.CC_COMMON_SHORT_SIZE
                )

            elif conv_type in (v23c.CONVERSION_TYPE_EXPO, v23c.CONVERSION_TYPE_LOGH):
                (
                    self.P1,
                    self.P2,
                    self.P3,
                    self.P4,
                    self.P5,
                    self.P6,
                    self.P7,
                ) = unpack_from("<7d", block, v23c.CC_COMMON_SHORT_SIZE)

            elif conv_type == v23c.CONVERSION_TYPE_TABX:
                nr = self.ref_param_nr

                size = v23c.CC_COMMON_BLOCK_SIZE + nr * 40

                if block_size == v23c.MAX_UINT16:
                    stream.seek(address)
                    raw_bytes = stream.read(size)
                    conversion = ChannelConversion(
                        raw_bytes=raw_bytes, stream=stream, address=address
                    )
                    conversion.block_len = size

                    for attr in get_fields(conversion):
                        setattr(self, attr, getattr(conversion, attr))

                    self.referenced_blocks = conversion.referenced_blocks

                else:

                    values = unpack_from(
                        "<" + "d32s" * nr, block, v23c.CC_COMMON_SHORT_SIZE
                    )

                    for i in range(nr):
                        (self[f"param_val_{i}"], self[f"text_{i}"]) = (
                            values[i * 2],
                            values[2 * i + 1],
                        )

            elif conv_type == v23c.CONVERSION_TYPE_RTABX:

                nr = self.ref_param_nr - 1

                size = v23c.CC_COMMON_BLOCK_SIZE + (nr + 1) * 20

                if block_size == v23c.MAX_UINT16:
                    stream.seek(address)
                    raw_bytes = stream.read(size)
                    conversion = ChannelConversion(
                        raw_bytes=raw_bytes, stream=stream, address=address
                    )
                    conversion.block_len = size

                    for attr in get_fields(conversion):
                        setattr(self, attr, getattr(conversion, attr))
                    self.referenced_blocks = conversion.referenced_blocks

                else:

                    (
                        self.default_lower,
                        self.default_upper,
                        self.default_addr,
                    ) = unpack_from("<2dI", block, v23c.CC_COMMON_SHORT_SIZE)

                    if self.default_addr:
                        self.referenced_blocks["default_addr"] = get_text_v3(
                            address=self.default_addr,
                            stream=stream,
                            mapped=mapped,
                            decode=False,
                        )
                    else:
                        self.referenced_blocks["default_addr"] = b""

                    values = unpack_from(
                        "<" + "2dI" * nr, block, v23c.CC_COMMON_SHORT_SIZE + 20
                    )

                    for i in range(nr):
                        (self[f"lower_{i}"], self[f"upper_{i}"], self[f"text_{i}"]) = (
                            values[i * 3],
                            values[3 * i + 1],
                            values[3 * i + 2],
                        )
                        if values[3 * i + 2]:
                            block = get_text_v3(
                                address=values[3 * i + 2],
                                stream=stream,
                                mapped=mapped,
                                decode=False,
                            )
                            self.referenced_blocks[f"text_{i}"] = block

                        else:
                            self.referenced_blocks[f"text_{i}"] = b""

            if self.id != b"CC":
                message = f'Expected "CC" block @{hex(address)} but found "{self.id}"'

                logger.exception(message)
                raise MdfException(message)

        else:

            self.address = 0
            self.id = b"CC"
            self.unit_field = kwargs.get("unit", b"")

            if kwargs["conversion_type"] == v23c.CONVERSION_TYPE_NONE:
                self.block_len = v23c.CC_COMMON_BLOCK_SIZE
                self.range_flag = kwargs.get("range_flag", 1)
                self.min_phy_value = kwargs.get("min_phy_value", 0)
                self.max_phy_value = kwargs.get("max_phy_value", 0)
                self.unit_field = kwargs.get("unit", ("\0" * 20).encode("latin-1"))
                self.conversion_type = v23c.CONVERSION_TYPE_NONE
                self.ref_param_nr = 0

            elif kwargs["conversion_type"] == v23c.CONVERSION_TYPE_LINEAR:
                self.block_len = v23c.CC_LIN_BLOCK_SIZE
                self.range_flag = kwargs.get("range_flag", 1)
                self.min_phy_value = kwargs.get("min_phy_value", 0)
                self.max_phy_value = kwargs.get("max_phy_value", 0)
                self.unit_field = kwargs.get("unit", ("\0" * 20).encode("latin-1"))
                self.conversion_type = v23c.CONVERSION_TYPE_LINEAR
                self.ref_param_nr = 2
                self.b = kwargs.get("b", 0)
                self.a = kwargs.get("a", 1)
                if not self.block_len == v23c.CC_LIN_BLOCK_SIZE:
                    self.CANapeHiddenExtra = kwargs["CANapeHiddenExtra"]

            elif kwargs["conversion_type"] in (
                v23c.CONVERSION_TYPE_POLY,
                v23c.CONVERSION_TYPE_RAT,
            ):
                self.block_len = v23c.CC_POLY_BLOCK_SIZE
                self.range_flag = kwargs.get("range_flag", 1)
                self.min_phy_value = kwargs.get("min_phy_value", 0)
                self.max_phy_value = kwargs.get("max_phy_value", 0)
                self.unit_field = kwargs.get("unit", ("\0" * 20).encode("latin-1"))
                self.conversion_type = kwargs["conversion_type"]
                self.ref_param_nr = 6
                self.P1 = kwargs.get("P1", 0)
                self.P2 = kwargs.get("P2", 0)
                self.P3 = kwargs.get("P3", 0)
                self.P4 = kwargs.get("P4", 0)
                self.P5 = kwargs.get("P5", 0)
                self.P6 = kwargs.get("P6", 0)

            elif kwargs["conversion_type"] in (
                v23c.CONVERSION_TYPE_EXPO,
                v23c.CONVERSION_TYPE_LOGH,
            ):
                self.block_len = v23c.CC_EXPO_BLOCK_SIZE
                self.range_flag = kwargs.get("range_flag", 1)
                self.min_phy_value = kwargs.get("min_phy_value", 0)
                self.max_phy_value = kwargs.get("max_phy_value", 0)
                self.unit_field = kwargs.get("unit", ("\0" * 20).encode("latin-1"))
                self.conversion_type = v23c.CONVERSION_TYPE_EXPO
                self.ref_param_nr = 7
                self.P1 = kwargs.get("P1", 0)
                self.P2 = kwargs.get("P2", 0)
                self.P3 = kwargs.get("P3", 0)
                self.P4 = kwargs.get("P4", 0)
                self.P5 = kwargs.get("P5", 0)
                self.P6 = kwargs.get("P6", 0)
                self.P7 = kwargs.get("P7", 0)

            elif kwargs["conversion_type"] == v23c.CONVERSION_TYPE_FORMULA:
                formula = kwargs["formula"]
                formula_len = len(formula)
                try:
                    self.formula = formula.decode("latin-1")
                    formula += b"\0"
                except AttributeError:
                    self.formula = formula
                    formula = formula.encode("latin-1") + b"\0"
                self.block_len = 46 + formula_len + 1
                self.range_flag = kwargs.get("range_flag", 1)
                self.min_phy_value = kwargs.get("min_phy_value", 0)
                self.max_phy_value = kwargs.get("max_phy_value", 0)
                self.unit_field = kwargs.get("unit", ("\0" * 20).encode("latin-1"))
                self.conversion_type = v23c.CONVERSION_TYPE_FORMULA
                self.ref_param_nr = formula_len
                self.formula_field = formula

            elif kwargs["conversion_type"] in (
                v23c.CONVERSION_TYPE_TABI,
                v23c.CONVERSION_TYPE_TAB,
            ):
                nr = kwargs["ref_param_nr"]
                self.block_len = v23c.CC_COMMON_BLOCK_SIZE + nr * 2 * 8
                self.range_flag = kwargs.get("range_flag", 1)
                self.min_phy_value = kwargs.get("min_phy_value", 0)
                self.max_phy_value = kwargs.get("max_phy_value", 0)
                self.unit_field = kwargs.get("unit", ("\0" * 20).encode("latin-1"))
                self.conversion_type = kwargs["conversion_type"]
                self.ref_param_nr = nr
                for i in range(nr):
                    self[f"raw_{i}"] = kwargs[f"raw_{i}"]
                    self[f"phys_{i}"] = kwargs[f"phys_{i}"]

            elif kwargs["conversion_type"] == v23c.CONVERSION_TYPE_TABX:
                nr = kwargs["ref_param_nr"]
                self.block_len = v23c.CC_COMMON_BLOCK_SIZE + 40 * nr
                self.range_flag = kwargs.get("range_flag", 0)
                self.min_phy_value = kwargs.get("min_phy_value", 0)
                self.max_phy_value = kwargs.get("max_phy_value", 0)
                self.unit_field = kwargs.get("unit", ("\0" * 20).encode("latin-1"))
                self.conversion_type = v23c.CONVERSION_TYPE_TABX
                self.ref_param_nr = nr

                for i in range(nr):
                    self[f"param_val_{i}"] = kwargs[f"param_val_{i}"]
                    self[f"text_{i}"] = kwargs[f"text_{i}"]

            elif kwargs["conversion_type"] == v23c.CONVERSION_TYPE_RTABX:
                nr = kwargs["ref_param_nr"]
                self.block_len = v23c.CC_COMMON_BLOCK_SIZE + 20 * nr
                self.range_flag = kwargs.get("range_flag", 0)
                self.min_phy_value = kwargs.get("min_phy_value", 0)
                self.max_phy_value = kwargs.get("max_phy_value", 0)
                self.unit_field = kwargs.get("unit", ("\0" * 20).encode("latin-1"))
                self.conversion_type = v23c.CONVERSION_TYPE_RTABX
                self.ref_param_nr = nr

                self.default_lower = 0
                self.default_upper = 0
                self.default_addr = 0
                key = "default_addr"
                if key in kwargs:
                    self.referenced_blocks[key] = kwargs[key]
                else:
                    self.referenced_blocks[key] = b""

                for i in range(nr - 1):
                    self[f"lower_{i}"] = kwargs[f"lower_{i}"]
                    self[f"upper_{i}"] = kwargs[f"upper_{i}"]
                    key = f"text_{i}"
                    self[key] = 0
                    self.referenced_blocks[key] = kwargs[key]
            else:
                message = (
                    f'Conversion type "{kwargs["conversion_type"]}" not implemented'
                )
                logger.exception(message)
                raise MdfException(message)

    def to_blocks(
        self,
        address: int,
        blocks: list[Any],
        defined_texts: dict[str, int],
        cc_map: dict[bytes, int],
    ) -> int:
        self.unit_field = self.unit.encode("latin-1", "ignore")[:19]

        if self.conversion_type == v23c.CONVERSION_TYPE_FORMULA:
            formula = self.formula
            if not formula.endswith("\0"):
                formula += "\0"
            self.formula_field = formula.encode("latin-1")
            self.block_len = v23c.CC_COMMON_BLOCK_SIZE + len(self.formula_field)

        for key, block in self.referenced_blocks.items():
            if block:
                if isinstance(block, ChannelConversion):
                    address = block.to_blocks(address, blocks, defined_texts, cc_map)
                    self[key] = block.address
                else:
                    text = block
                    if text in defined_texts:
                        self[key] = defined_texts[text]
                    else:
                        block = TextBlock(text=text)
                        defined_texts[text] = address
                        blocks.append(block)
                        self[key] = address
                        address += block.block_len
            else:
                self[key] = 0

        bts = bytes(self)
        if bts in cc_map:
            self.address = cc_map[bts]
        else:
            blocks.append(bts)
            self.address = address
            cc_map[bts] = address
            size = self.block_len
            address += len(bts)

        return address

    def metadata(self, indent: str = "") -> str:
        conv = self.conversion_type
        if conv == v23c.CONVERSION_TYPE_NONE:
            keys = v23c.KEYS_CONVERSION_NONE
        elif conv == v23c.CONVERSION_TYPE_FORMULA:
            keys = v23c.KEYS_CONVERSION_FORMULA
        elif conv == v23c.CONVERSION_TYPE_LINEAR:
            keys = v23c.KEYS_CONVERSION_LINEAR
            if not self.block_len == v23c.CC_LIN_BLOCK_SIZE:
                keys += ("CANapeHiddenExtra",)

                nr = self.ref_param_nr
                new_keys = []
                for i in range(nr):
                    new_keys.append(f"param_val_{i}")
                    new_keys.append(f"text_{i}")
                keys += tuple(new_keys)

        elif conv in (v23c.CONVERSION_TYPE_POLY, v23c.CONVERSION_TYPE_RAT):
            keys = v23c.KEYS_CONVERSION_POLY_RAT
        elif conv in (v23c.CONVERSION_TYPE_EXPO, v23c.CONVERSION_TYPE_LOGH):
            keys = v23c.KEYS_CONVERSION_EXPO_LOGH
        elif conv in (v23c.CONVERSION_TYPE_TABI, v23c.CONVERSION_TYPE_TAB):
            nr = self.ref_param_nr
            keys = list(v23c.KEYS_CONVERSION_NONE)
            for i in range(nr):
                keys.append(f"raw_{i}")
                keys.append(f"phys_{i}")
        elif conv == v23c.CONVERSION_TYPE_RTABX:
            nr = self.ref_param_nr
            keys = list(v23c.KEYS_CONVERSION_NONE)
            keys += ["default_lower", "default_upper", "default_addr"]
            for i in range(nr - 1):
                keys.append(f"lower_{i}")
                keys.append(f"upper_{i}")
                keys.append(f"text_{i}")
        elif conv == v23c.CONVERSION_TYPE_TABX:
            nr = self.ref_param_nr
            keys = list(v23c.KEYS_CONVERSION_NONE)
            for i in range(nr):
                keys.append(f"param_val_{i}")
                keys.append(f"text_{i}")

        max_len = max(len(key) for key in keys)
        template = f"{{: <{max_len}}}: {{}}"

        metadata = []
        lines = f"""
address: {hex(self.address)}

""".split(
            "\n"
        )

        for key in keys:
            val = getattr(self, key)
            if key.endswith("addr") or key.startswith("text_") and isinstance(val, int):
                lines.append(template.format(key, hex(val)))
            elif isinstance(val, float):
                lines.append(template.format(key, round(val, 6)))
            else:
                if isinstance(val, bytes):
                    lines.append(template.format(key, val.strip(b"\0")))
                else:
                    lines.append(template.format(key, val))

            if key == "conversion_type":
                lines[
                    -1
                ] += f" [{v23c.CONVERSION_TYPE_TO_STRING[self.conversion_type]}]"
            elif self.referenced_blocks and key in self.referenced_blocks:
                val = self.referenced_blocks[key]
                if isinstance(val, bytes):
                    lines[-1] += f" (= {str(val)[1:]})"
                else:
                    lines[-1] += f" (= CCBLOCK @ {hex(val.address)})"

        if self.referenced_blocks:
            max_len = max(len(key) for key in self.referenced_blocks)
            template = f"{{: <{max_len}}}: {{}}"

            lines.append("")
            lines.append("Referenced blocks:")
            for key, block in self.referenced_blocks.items():
                if isinstance(block, ChannelConversion):
                    lines.append(template.format(key, ""))
                    lines.extend(block.metadata(indent + "    ").split("\n"))
                else:
                    lines.append(template.format(key, str(block)[1:]))

        for line in lines:
            if not line:
                metadata.append(line)
            else:
                for wrapped_line in wrap(
                    line, initial_indent=indent, subsequent_indent=indent, width=120
                ):
                    metadata.append(wrapped_line)

        return "\n".join(metadata)

    def convert(self, values):
        conversion_type = self.conversion_type

        if conversion_type == v23c.CONVERSION_TYPE_NONE:
            pass

        elif conversion_type == v23c.CONVERSION_TYPE_LINEAR:
            a = self.a
            b = self.b
            if (a, b) != (1, 0):
                values = values * a
                if b:
                    values += b

        elif conversion_type in (v23c.CONVERSION_TYPE_TABI, v23c.CONVERSION_TYPE_TAB):
            nr = self.ref_param_nr

            raw_vals = [self[f"raw_{i}"] for i in range(nr)]
            raw_vals = np.array(raw_vals)
            phys = [self[f"phys_{i}"] for i in range(nr)]
            phys = np.array(phys)

            if conversion_type == v23c.CONVERSION_TYPE_TABI:
                values = np.interp(values, raw_vals, phys)
            else:
                dim = raw_vals.shape[0]

                inds = np.searchsorted(raw_vals, values)

                inds[inds >= dim] = dim - 1

                inds2 = inds - 1
                inds2[inds2 < 0] = 0

                cond = np.abs(values - raw_vals[inds]) >= np.abs(
                    values - raw_vals[inds2]
                )

                values = np.where(cond, phys[inds2], phys[inds])

        elif conversion_type == v23c.CONVERSION_TYPE_TABX:
            nr = self.ref_param_nr
            raw_vals = [self[f"param_val_{i}"] for i in range(nr)]
            raw_vals = np.array(raw_vals)
            phys = [self[f"text_{i}"] for i in range(nr)]
            phys = np.array(phys)

            x = sorted(zip(raw_vals, phys))
            raw_vals = np.array([e[0] for e in x], dtype="<i8")
            phys = np.array([e[1] for e in x])

            default = b""

            idx1 = np.searchsorted(raw_vals, values, side="right") - 1
            idx2 = np.searchsorted(raw_vals, values, side="left")

            idx = np.argwhere(idx1 != idx2).flatten()

            new_values = np.zeros(
                len(values), dtype=max(phys.dtype, np.array([default]).dtype)
            )

            new_values[idx] = default
            idx = np.argwhere(idx1 == idx2).flatten()
            if len(idx):
                new_values[idx] = phys[idx1[idx]]

            values = new_values

        elif conversion_type == v23c.CONVERSION_TYPE_RTABX:
            nr = self.ref_param_nr - 1

            phys = []
            for i in range(nr):
                value = self.referenced_blocks[f"text_{i}"]
                phys.append(value)

            phys = np.array(phys)

            default = self.referenced_blocks["default_addr"]

            if b"{X}" in default:
                default = default.decode("latin-1").replace("{X}", "X").split('"')[1]
                partial_conversion = True
            else:
                partial_conversion = False

            lower = np.array([self[f"lower_{i}"] for i in range(nr)])
            upper = np.array([self[f"upper_{i}"] for i in range(nr)])

            idx1 = np.searchsorted(lower, values, side="right") - 1
            idx2 = np.searchsorted(upper, values, side="left")

            idx = np.argwhere(idx1 != idx2).flatten()

            if partial_conversion and len(idx):
                X = values[idx]
                new_values = np.zeros(len(values), dtype=np.float64)
                new_values[idx] = evaluate(default)

                idx = np.argwhere(idx1 == idx2).flatten()
                if len(idx):
                    new_values[idx] = np.nan
                values = new_values

            else:
                if len(idx):
                    new_values = np.zeros(
                        len(values), dtype=max(phys.dtype, np.array([default]).dtype)
                    )
                    new_values[idx] = default

                    idx = np.argwhere(idx1 == idx2).flatten()
                    if len(idx):
                        new_values[idx] = phys[idx1[idx]]

                    values = new_values
                else:
                    values = phys[idx1]

        elif conversion_type in (v23c.CONVERSION_TYPE_EXPO, v23c.CONVERSION_TYPE_LOGH):
            # pylint: disable=C0103

            if conversion_type == v23c.CONVERSION_TYPE_EXPO:
                func = np.log
            else:
                func = np.exp
            P1 = self.P1
            P2 = self.P2
            P3 = self.P3
            P4 = self.P4
            P5 = self.P5
            P6 = self.P6
            P7 = self.P7
            if P4 == 0:
                values = func(((values - P7) * P6 - P3) / P1) / P2
            elif P1 == 0:
                values = func((P3 / (values - P7) - P6) / P4) / P5
            else:
                message = f"wrong conversion {conversion_type}"
                raise ValueError(message)

        elif conversion_type == v23c.CONVERSION_TYPE_RAT:
            # pylint: disable=unused-variable,C0103

            P1 = self.P1
            P2 = self.P2
            P3 = self.P3
            P4 = self.P4
            P5 = self.P5
            P6 = self.P6

            X = values
            if (P1, P4, P5, P6) == (0, 0, 0, 1):
                if (P2, P3) != (1, 0):
                    values = values * P2
                    if P3:
                        values += P3
            elif (P3, P4, P5, P6) == (0, 0, 1, 0):
                if (P1, P2) != (1, 0):
                    values = values * P1
                    if P2:
                        values += P2
            else:
                try:
                    values = evaluate(v23c.RAT_CONV_TEXT)
                except TypeError:
                    values = (P1 * X**2 + P2 * X + P3) / (P4 * X**2 + P5 * X + P6)

        elif conversion_type == v23c.CONVERSION_TYPE_POLY:
            # pylint: disable=unused-variable,C0103

            P1 = self.P1
            P2 = self.P2
            P3 = self.P3
            P4 = self.P4
            P5 = self.P5
            P6 = self.P6

            X = values

            coefs = (P2, P3, P5, P6)
            if coefs == (0, 0, 0, 0):
                if P1 != P4:
                    try:
                        values = evaluate(v23c.POLY_CONV_SHORT_TEXT)
                    except TypeError:
                        values = P4 * X / P1
            else:

                try:
                    values = evaluate(v23c.POLY_CONV_LONG_TEXT)
                except TypeError:
                    values = (P2 - (P4 * (X - P5 - P6))) / (P3 * (X - P5 - P6) - P1)

        elif conversion_type == v23c.CONVERSION_TYPE_FORMULA:
            # pylint: disable=unused-variable,C0103

            X1 = values
            try:
                values = evaluate(self.formula)
            except:
                values = evaluate3(self.formula)

        return values

    def __getitem__(self, item: str) -> Any:
        return self.__getattribute__(item)

    def __setitem__(self, item: str, value: Any) -> None:
        self.__setattr__(item, value)

    def __bytes__(self) -> bytes:
        conv = self.conversion_type

        # compute the fmt
        if conv == v23c.CONVERSION_TYPE_NONE:
            fmt = v23c.FMT_CONVERSION_COMMON
        elif conv == v23c.CONVERSION_TYPE_FORMULA:
            fmt = v23c.FMT_CONVERSION_FORMULA.format(
                self.block_len - v23c.CC_COMMON_BLOCK_SIZE
            )
        elif conv == v23c.CONVERSION_TYPE_LINEAR:
            fmt = v23c.FMT_CONVERSION_LINEAR
            if not self.block_len == v23c.CC_LIN_BLOCK_SIZE:
                fmt += f"{self.block_len - v23c.CC_LIN_BLOCK_SIZE}s"
        elif conv in (v23c.CONVERSION_TYPE_POLY, v23c.CONVERSION_TYPE_RAT):
            fmt = v23c.FMT_CONVERSION_POLY_RAT
        elif conv in (v23c.CONVERSION_TYPE_EXPO, v23c.CONVERSION_TYPE_LOGH):
            fmt = v23c.FMT_CONVERSION_EXPO_LOGH
        elif conv in (v23c.CONVERSION_TYPE_TABI, v23c.CONVERSION_TYPE_TAB):
            nr = self.ref_param_nr
            fmt = v23c.FMT_CONVERSION_COMMON + f"{2*nr}d"
        elif conv == v23c.CONVERSION_TYPE_RTABX:
            nr = self.ref_param_nr
            fmt = v23c.FMT_CONVERSION_COMMON + "2dI" * nr
        elif conv == v23c.CONVERSION_TYPE_TABX:
            nr = self.ref_param_nr
            fmt = v23c.FMT_CONVERSION_COMMON + "d32s" * nr

        if conv == v23c.CONVERSION_TYPE_NONE:
            keys = v23c.KEYS_CONVERSION_NONE
        elif conv == v23c.CONVERSION_TYPE_FORMULA:
            keys = v23c.KEYS_CONVERSION_FORMULA
        elif conv == v23c.CONVERSION_TYPE_LINEAR:
            keys = v23c.KEYS_CONVERSION_LINEAR
            if not self.block_len == v23c.CC_LIN_BLOCK_SIZE:
                keys += ("CANapeHiddenExtra",)
        elif conv in (v23c.CONVERSION_TYPE_POLY, v23c.CONVERSION_TYPE_RAT):
            keys = v23c.KEYS_CONVERSION_POLY_RAT
        elif conv in (v23c.CONVERSION_TYPE_EXPO, v23c.CONVERSION_TYPE_LOGH):
            keys = v23c.KEYS_CONVERSION_EXPO_LOGH
        elif conv in (v23c.CONVERSION_TYPE_TABI, v23c.CONVERSION_TYPE_TAB):
            nr = self.ref_param_nr
            keys = list(v23c.KEYS_CONVERSION_NONE)
            for i in range(nr):
                keys.append(f"raw_{i}")
                keys.append(f"phys_{i}")
        elif conv == v23c.CONVERSION_TYPE_RTABX:
            nr = self.ref_param_nr
            keys = list(v23c.KEYS_CONVERSION_NONE)
            keys += ["default_lower", "default_upper", "default_addr"]
            for i in range(nr - 1):
                keys.append(f"lower_{i}")
                keys.append(f"upper_{i}")
                keys.append(f"text_{i}")
        elif conv == v23c.CONVERSION_TYPE_TABX:
            nr = self.ref_param_nr
            keys = list(v23c.KEYS_CONVERSION_NONE)
            for i in range(nr):
                keys.append(f"param_val_{i}")
                keys.append(f"text_{i}")

        if self.block_len > v23c.MAX_UINT16:
            self.block_len = v23c.MAX_UINT16
        result = pack(fmt, *[self[key] for key in keys])
        return result

    def __str__(self) -> str:
        fields = []
        for attr in dir(self):
            if attr[:2] + attr[-2:] == "____":
                continue
            try:
                if callable(getattr(self, attr)):
                    continue
                fields.append(f"{attr}:{getattr(self, attr)}")
            except AttributeError:
                continue
        return f"ChannelConversion (referenced blocks: {self.referenced_blocks}, address: {hex(self.address)}, fields: {fields})" class ChannelDependency:
    """CDBLOCK class

    CDBLOCK fields

    * ``id`` - bytes : block ID; always b'CD'
    * ``block_len`` - int : block bytes size
    * ``dependency_type`` - int : integer code for dependency type
    * ``sd_nr`` - int : total number of signals dependencies
    * ``dg_<N>`` - address of data group block (DGBLOCK) of N-th
      signal dependency
    * ``dg_<N>`` - address of channel group block (CGBLOCK) of N-th
      signal dependency
    * ``dg_<N>`` - address of channel block (CNBLOCK) of N-th
      signal dependency
    * ``dim_<K>`` - int : Optional size of dimension *K* for N-dimensional
      dependency

    Other attributes
    * ``address`` - int : block address inside mdf file
    * ``referenced_channels`` - list : list of (group index, channel index) pairs

    Parameters
    ----------
    stream : file handle
        mdf file handle
    address : int
        block address inside mdf file
    for dynamically created objects :
        see the key-value pairs



    """

    def __init__(self, **kwargs) -> None:
        super().__init__()

        self.referenced_channels = []

        try:
            stream = kwargs["stream"]
            self.address = address = kwargs["address"]
            stream.seek(address)

            (self.id, self.block_len, self.dependency_type, self.sd_nr) = unpack(
                "<2s3H", stream.read(8)
            )

            links_size = 3 * 4 * self.sd_nr
            links = unpack("<{}I".format(3 * self.sd_nr), stream.read(links_size))

            for i in range(self.sd_nr):
                self[f"dg_{i}"] = links[3 * i]
                self[f"cg_{i}"] = links[3 * i + 1]
                self[f"ch_{i}"] = links[3 * i + 2]

            optional_dims_nr = (self.block_len - 8 - links_size) // 2
            if optional_dims_nr:
                dims = unpack(
                    f"<{optional_dims_nr}H", stream.read(optional_dims_nr * 2)
                )
                for i, dim in enumerate(dims):
                    self[f"dim_{i}"] = dim

            if self.id != b"CD":
                message = f'Expected "CD" block @{hex(address)} but found "{self.id}"'

                logger.exception(message)
                raise MdfException(message)

        except KeyError:
            sd_nr = kwargs["sd_nr"]
            self.id = b"CD"
            self.block_len = 8 + 3 * 4 * sd_nr
            self.dependency_type = 1
            self.sd_nr = sd_nr
            for i in range(sd_nr):
                self[f"dg_{i}"] = 0
                self[f"cg_{i}"] = 0
                self[f"ch_{i}"] = 0
            i = 0
            while True:
                try:
                    self[f"dim_{i}"] = kwargs[f"dim_{i}"]
                    i += 1
                except KeyError:
                    break
            if i:
                self.dependency_type = 256 + i
                self.block_len += 2 * i

    def __getitem__(self, item: str) -> Any:
        return self.__getattribute__(item)

    def __setitem__(self, item: str, value: Any) -> None:
        self.__setattr__(item, value)

    def __bytes__(self) -> bytes:
        fmt = f"<2s3H{self.sd_nr * 3}I"
        keys = ("id", "block_len", "dependency_type", "sd_nr")
        for i in range(self.sd_nr):
            keys += (f"dg_{i}", f"cg_{i}", f"ch_{i}")
        links_size = 3 * 4 * self.sd_nr
        option_dims_nr = (self.block_len - 8 - links_size) // 2
        if option_dims_nr:
            fmt += f"{option_dims_nr}H"
            keys += tuple(f"dim_{i}" for i in range(option_dims_nr))
        result = pack(fmt, *[self[key] for key in keys])
        return result class ChannelExtension:
    """CEBLOCK class

    CEBLOCK has the following common fields

    * ``id`` - bytes : block ID; always b'CE'
    * ``block_len`` - int : block bytes size
    * ``type`` - int : extension type identifier

    CEBLOCK has the following specific fields

    * for DIM block

        * ``module_nr`` - int: module number
        * ``module_address`` - int : module address
        * ``description`` - bytes : module description
        * ``ECU_identification`` - bytes : identification of ECU
        * ``reserved0`` - bytes : reserved bytes

    * for Vector CAN block

        * ``CAN_id`` - int : CAN message ID
        * ``CAN_ch_index`` - int : index of CAN channel
        * ``message_name`` - bytes : message name
        * ``sender_name`` - btyes : sender name
        * ``reserved0`` - bytes : reserved bytes

    Other attributes

    * ``address`` - int : block address inside mdf file
    * ``comment`` - str : extension comment
    * ``name`` - str : extension name
    * ``path`` - str : extension path

    Parameters
    ----------
    stream : file handle
        mdf file handle
    address : int
        block address inside mdf file
    for dynamically created objects :
        see the key-value pairs

    """

    __slots__ = (
        "address",
        "name",
        "path",
        "comment",
        "id",
        "block_len",
        "type",
        "module_nr",
        "module_address",
        "description",
        "ECU_identification",
        "reserved0",
        "CAN_id",
        "CAN_ch_index",
        "message_name",
        "sender_name",
    )

    def __init__(self, **kwargs) -> None:
        super().__init__()

        self.name = self.path = self.comment = ""

        if "stream" in kwargs:
            stream = kwargs["stream"]
            try:

                (self.id, self.block_len, self.type) = SOURCE_COMMON_uf(
                    kwargs["raw_bytes"]
                )
                if self.type == v23c.SOURCE_ECU:
                    (
                        self.module_nr,
                        self.module_address,
                        self.description,
                        self.ECU_identification,
                        self.reserved0,
                    ) = SOURCE_EXTRA_ECU_uf(kwargs["raw_bytes"], 6)
                elif self.type == v23c.SOURCE_VECTOR:
                    (
                        self.CAN_id,
                        self.CAN_ch_index,
                        self.message_name,
                        self.sender_name,
                        self.reserved0,
                    ) = SOURCE_EXTRA_VECTOR_uf(kwargs["raw_bytes"], 6)

                self.address = kwargs.get("address", 0)
            except KeyError:
                if kwargs.get("mapped", False):
                    self.address = address = kwargs["address"]

                    (self.id, self.block_len, self.type) = SOURCE_COMMON_uf(
                        stream, address
                    )

                    if self.type == v23c.SOURCE_ECU:
                        (
                            self.module_nr,
                            self.module_address,
                            self.description,
                            self.ECU_identification,
                            self.reserved0,
                        ) = SOURCE_EXTRA_ECU_uf(stream, address + 6)
                    elif self.type == v23c.SOURCE_VECTOR:
                        (
                            self.CAN_id,
                            self.CAN_ch_index,
                            self.message_name,
                            self.sender_name,
                            self.reserved0,
                        ) = SOURCE_EXTRA_VECTOR_uf(stream, address + 6)
                else:
                    self.address = address = kwargs["address"]
                    stream.seek(address)
                    block = stream.read(v23c.CE_BLOCK_SIZE)
                    (self.id, self.block_len, self.type) = SOURCE_COMMON_uf(block)

                    if self.type == v23c.SOURCE_ECU:
                        (
                            self.module_nr,
                            self.module_address,
                            self.description,
                            self.ECU_identification,
                            self.reserved0,
                        ) = SOURCE_EXTRA_ECU_uf(block, 6)
                    elif self.type == v23c.SOURCE_VECTOR:
                        (
                            self.CAN_id,
                            self.CAN_ch_index,
                            self.message_name,
                            self.sender_name,
                            self.reserved0,
                        ) = SOURCE_EXTRA_VECTOR_uf(block, 6)

            if self.id != b"CE":
                message = f'Expected "CE" block @{hex(address)} but found "{self.id}"'

                logger.exception(message)
                raise MdfException(message)

        else:

            self.address = 0
            self.id = b"CE"
            self.block_len = kwargs.get("block_len", v23c.CE_BLOCK_SIZE)
            self.type = kwargs.get("type", 2)
            if self.type == v23c.SOURCE_ECU:
                self.module_nr = kwargs.get("module_nr", 0)
                self.module_address = kwargs.get("module_address", 0)
                self.description = kwargs.get("description", b"\0")
                self.ECU_identification = kwargs.get("ECU_identification", b"\0")
                self.reserved0 = kwargs.get("reserved0", b"\0")
            elif self.type == v23c.SOURCE_VECTOR:
                self.CAN_id = kwargs.get("CAN_id", 0)
                self.CAN_ch_index = kwargs.get("CAN_ch_index", 0)
                self.message_name = kwargs.get("message_name", b"\0")
                self.sender_name = kwargs.get("sender_name", b"\0")
                self.reserved0 = kwargs.get("reserved0", b"\0")

        if self.type == v23c.SOURCE_ECU:
            self.path = self.ECU_identification.decode("latin-1").strip(" \t\n\r\0")
            self.name = self.description.decode("latin-1").strip(" \t\n\r\0")
            self.comment = (
                f"Module number={self.module_nr} @ address={self.module_address}"
            )
        else:
            self.path = self.sender_name.decode("latin-1").strip(" \t\n\r\0")
            self.name = self.message_name.decode("latin-1").strip(" \t\n\r\0")
            self.comment = (
                f"Message ID={hex(self.CAN_id)} on CAN bus {self.CAN_ch_index}"
            )

    def to_blocks(
        self,
        address: int,
        blocks: list[Any],
        defined_texts: dict[str, int],
        cc_map: dict[bytes, int],
    ) -> int:

        if self.type == v23c.SOURCE_ECU:
            self.ECU_identification = self.path.encode("latin-1")[:31]
            self.description = self.name.encode("latin-1")[:79]
        else:
            self.sender_name = self.path.encode("latin-1")[:35]
            self.message_name = self.name.encode("latin-1")[:35]

        bts = bytes(self)
        if bts in cc_map:
            self.address = cc_map[bts]
        else:
            blocks.append(bts)
            self.address = address
            cc_map[bts] = address
            address += self.block_len

        return address

    def __getitem__(self, item: str) -> Any:
        return self.__getattribute__(item)

    def __setitem__(self, item: str, value: Any) -> None:
        self.__setattr__(item, value)

    def metadata(self) -> str:
        if self.type == v23c.SOURCE_ECU:
            keys = (
                "id",
                "block_len",
                "type",
                "module_nr",
                "module_address",
                "description",
                "ECU_identification",
                "reserved0",
            )
        else:
            keys = (
                "id",
                "block_len",
                "type",
                "CAN_id",
                "CAN_ch_index",
                "message_name",
                "sender_name",
                "reserved0",
            )
        max_len = max(len(key) for key in keys)
        template = f"{{: <{max_len}}}: {{}}"

        metadata = []
        lines = f"""
address: {hex(self.address)}

""".split(
            "\n"
        )

        for key in keys:
            val = getattr(self, key)
            if key.endswith("addr") or key.startswith("text_"):
                lines.append(template.format(key, hex(val)))
            elif isinstance(val, float):
                lines.append(template.format(key, round(val, 6)))
            else:
                if isinstance(val, bytes):
                    lines.append(template.format(key, val.strip(b"\0")))
                else:
                    lines.append(template.format(key, val))

            if key == "type":
                lines[-1] += f" = {v23c.SOURCE_TYPE_TO_STRING[self.type]}"
            elif key == "bus_type":
                lines[-1] += f" = {v23c.BUS_TYPE_TO_STRING[self.bus_type]}"

        for line in lines:
            if not line:
                metadata.append(line)
            else:
                for wrapped_line in wrap(line, width=120):
                    metadata.append(wrapped_line)

        return "\n".join(metadata)

    def __bytes__(self) -> bytes:
        typ = self.type
        if typ == v23c.SOURCE_ECU:
            return v23c.SOURCE_ECU_p(
                self.id,
                self.block_len,
                self.type,
                self.module_nr,
                self.module_address,
                self.description,
                self.ECU_identification,
                self.reserved0,
            )
        else:
            return v23c.SOURCE_VECTOR_p(
                self.id,
                self.block_len,
                self.type,
                self.CAN_id,
                self.CAN_ch_index,
                self.message_name,
                self.sender_name,
                self.reserved0,
            )

    def __str__(self) -> str:
        fields = []
        for attr in dir(self):
            if attr[:2] + attr[-2:] == "____":
                continue
            try:
                if callable(getattr(self, attr)):
                    continue
                fields.append(f"{attr}:{getattr(self, attr)}")
            except AttributeError:
                continue
        return f"ChannelExtension (name: {self.name}, path: {self.path}, comment: {self.comment}, address: {hex(self.address)}, fields: {fields})" class ChannelGroup:
    """CGBLOCK class

    CGBLOCK fields

    * ``id`` - bytes : block ID; always b'CG'
    * ``block_len`` - int : block bytes size
    * ``next_cg_addr`` - int : next CGBLOCK address
    * ``first_ch_addr`` - int : address of first channel block (CNBLOCK)
    * ``comment_addr`` - int : address of TXBLOCK that contains the channel
      group comment
    * ``record_id`` - int : record ID used as identifier for a record if
      the DGBLOCK defines a number of record IDs > 0 (unsorted group)
    * ``ch_nr`` - int : number of channels
    * ``samples_byte_nr`` - int : size of data record in bytes without
      record ID
    * ``cycles_nr`` - int : number of cycles (records) of this type in the data
      block
    * ``sample_reduction_addr`` - int : address to first sample reduction
      block

    Other attributes

    * ``address`` - int : block address inside mdf file
    * ``comment`` - str : channel group comment

    Parameters
    ----------
    stream : file handle
        mdf file handle
    address : int
        block address inside mdf file
    for dynamically created objects :
        see the key-value pairs


    Examples
    --------
    >>> with open('test.mdf', 'rb') as mdf:
    ...     cg1 = ChannelGroup(stream=mdf, address=0xBA52)
    >>> cg2 = ChannelGroup(sample_bytes_nr=32)
    >>> hex(cg1.address)
    0xBA52
    >>> cg1['id']
    b'CG'

    """

    __slots__ = (
        "address",
        "comment",
        "id",
        "block_len",
        "next_cg_addr",
        "first_ch_addr",
        "comment_addr",
        "record_id",
        "ch_nr",
        "samples_byte_nr",
        "cycles_nr",
        "sample_reduction_addr",
    )

    def __init__(self, **kwargs) -> None:
        super().__init__()
        self.comment = ""

        try:
            stream = kwargs["stream"]
            mapped = kwargs.get("mapped", False)
            self.address = address = kwargs["address"]

            if mapped:
                (
                    self.id,
                    self.block_len,
                    self.next_cg_addr,
                    self.first_ch_addr,
                    self.comment_addr,
                    self.record_id,
                    self.ch_nr,
                    self.samples_byte_nr,
                    self.cycles_nr,
                ) = v23c.CHANNEL_GROUP_uf(stream, address)
                if self.block_len == v23c.CG_POST_330_BLOCK_SIZE:
                    # sample reduction blocks are not yet used
                    self.sample_reduction_addr = 0
            else:
                stream.seek(address)
                block = stream.read(v23c.CG_PRE_330_BLOCK_SIZE)

                (
                    self.id,
                    self.block_len,
                    self.next_cg_addr,
                    self.first_ch_addr,
                    self.comment_addr,
                    self.record_id,
                    self.ch_nr,
                    self.samples_byte_nr,
                    self.cycles_nr,
                ) = unpack(v23c.FMT_CHANNEL_GROUP, block)
                if self.block_len == v23c.CG_POST_330_BLOCK_SIZE:
                    # sample reduction blocks are not yet used
                    self.sample_reduction_addr = 0
            if self.id != b"CG":
                message = f'Expected "CG" block @{hex(address)} but found "{self.id}"'

                raise MdfException(message.format(self.id))
            if self.comment_addr:
                self.comment = get_text_v3(
                    address=self.comment_addr, stream=stream, mapped=mapped
                )
        except KeyError:
            self.address = 0
            self.id = b"CG"
            self.block_len = kwargs.get("block_len", v23c.CG_PRE_330_BLOCK_SIZE)
            self.next_cg_addr = kwargs.get("next_cg_addr", 0)
            self.first_ch_addr = kwargs.get("first_ch_addr", 0)
            self.comment_addr = kwargs.get("comment_addr", 0)
            self.record_id = kwargs.get("record_id", 1)
            self.ch_nr = kwargs.get("ch_nr", 0)
            self.samples_byte_nr = kwargs.get("samples_byte_nr", 0)
            self.cycles_nr = kwargs.get("cycles_nr", 0)
            if self.block_len == v23c.CG_POST_330_BLOCK_SIZE:
                self.sample_reduction_addr = 0

    def to_blocks(
        self,
        address: int,
        blocks: list[Any],
        defined_texts: dict[str, int],
        si_map: dict[bytes, int],
    ) -> int:
        key = "comment_addr"
        text = self.comment
        if text:
            if text in defined_texts:
                self[key] = defined_texts[text]
            else:
                tx_block = TextBlock(text=text)
                self[key] = address
                defined_texts[text] = address
                tx_block.address = address
                address += tx_block.block_len
                blocks.append(tx_block)
        else:
            self[key] = 0

        blocks.append(self)
        self.address = address
        address += self.block_len

        return address

    def __getitem__(self, item: str) -> Any:
        return self.__getattribute__(item)

    def __setitem__(self, item: str, value: Any) -> None:
        self.__setattr__(item, value)

    def __bytes__(self) -> bytes:
        if self.block_len == v23c.CG_POST_330_BLOCK_SIZE:
            return (
                v23c.CHANNEL_GROUP_p(
                    self.id,
                    self.block_len,
                    self.next_cg_addr,
                    self.first_ch_addr,
                    self.comment_addr,
                    self.record_id,
                    self.ch_nr,
                    self.samples_byte_nr,
                    self.cycles_nr,
                )
                + b"\0" * 4
            )
        else:
            return v23c.CHANNEL_GROUP_p(
                self.id,
                self.block_len,
                self.next_cg_addr,
                self.first_ch_addr,
                self.comment_addr,
                self.record_id,
                self.ch_nr,
                self.samples_byte_nr,
                self.cycles_nr,
            )

    def metadata(self) -> str:
        keys = (
            "id",
            "block_len",
            "next_cg_addr",
            "first_ch_addr",
            "comment_addr",
            "record_id",
            "ch_nr",
            "samples_byte_nr",
            "cycles_nr",
            "sample_reduction_addr",
        )

        max_len = max(len(key) for key in keys)
        template = f"{{: <{max_len}}}: {{}}"

        metadata = []
        lines = f"""
address: {hex(self.address)}
comment: {self.comment}

""".split(
            "\n"
        )

        for key in keys:
            if not hasattr(self, key):
                continue
            val = getattr(self, key)
            if key.endswith("addr") or key.startswith("text_"):
                lines.append(template.format(key, hex(val)))
            elif isinstance(val, float):
                lines.append(template.format(key, round(val, 6)))
            else:
                if isinstance(val, bytes):
                    lines.append(template.format(key, val.strip(b"\0")))
                else:
                    lines.append(template.format(key, val))
        for line in lines:
            if not line:
                metadata.append(line)
            else:
                for wrapped_line in wrap(line, width=120):
                    metadata.append(wrapped_line)

        return "\n".join(metadata) class DataGroup:
    """DGBLOCK class

    DGBLOCK fields

    * ``id`` - bytes : block ID; always b'DG'
    * ``block_len`` - int : block bytes size
    * ``next_dg_addr`` - int : next DGBLOCK address
    * ``first_cg_addr`` - int : address of first channel group block (CGBLOCK)
    * ``trigger_addr`` - int : address of trigger block (TRBLOCK)
    * ``data_block_addr`` - address of data block
    * ``cg_nr`` - int : number of channel groups
    * ``record_id_len`` - int : number of record IDs in the data block
    * ``reserved0`` - bytes : reserved bytes

    Other attributes

    * ``address`` - int : block address inside mdf file

    Parameters
    ----------
    stream : file handle
        mdf file handle
    address : int
        block address inside mdf file
    for dynamically created objects :
        see the key-value pairs


    """

    __slots__ = (
        "address",
        "id",
        "block_len",
        "next_dg_addr",
        "first_cg_addr",
        "trigger_addr",
        "data_block_addr",
        "cg_nr",
        "record_id_len",
        "reserved0",
    )

    def __init__(self, **kwargs) -> None:
        super().__init__()

        try:
            stream = kwargs["stream"]
            mapped = kwargs.get("mapped", False)
            self.address = address = kwargs["address"]
            if mapped:
                block = stream.read(v23c.DG_PRE_320_BLOCK_SIZE)

                (
                    self.id,
                    self.block_len,
                    self.next_dg_addr,
                    self.first_cg_addr,
                    self.trigger_addr,
                    self.data_block_addr,
                    self.cg_nr,
                    self.record_id_len,
                ) = v23c.DATA_GROUP_PRE_320_uf(stream, address)

                if self.block_len == v23c.DG_POST_320_BLOCK_SIZE:
                    self.reserved0 = stream[
                        address
                        + v23c.DG_PRE_320_BLOCK_SIZE : address
                        + v23c.DG_POST_320_BLOCK_SIZE
                    ]
            else:
                stream.seek(address)
                block = stream.read(v23c.DG_PRE_320_BLOCK_SIZE)

                (
                    self.id,
                    self.block_len,
                    self.next_dg_addr,
                    self.first_cg_addr,
                    self.trigger_addr,
                    self.data_block_addr,
                    self.cg_nr,
                    self.record_id_len,
                ) = v23c.DATA_GROUP_PRE_320_u(block)

                if self.block_len == v23c.DG_POST_320_BLOCK_SIZE:
                    self.reserved0 = stream.read(4)

            if self.id != b"DG":
                message = f'Expected "DG" block @{hex(address)} but found "{self.id}"'

                logger.exception(message)
                raise MdfException(message)

        except KeyError:
            self.address = kwargs.get("address", 0)
            self.id = b"DG"
            self.block_len = kwargs.get("block_len", v23c.DG_PRE_320_BLOCK_SIZE)
            self.next_dg_addr = kwargs.get("next_dg_addr", 0)
            self.first_cg_addr = kwargs.get("first_cg_addr", 0)
            self.trigger_addr = kwargs.get("comment_addr", 0)
            self.data_block_addr = kwargs.get("data_block_addr", 0)
            self.cg_nr = kwargs.get("cg_nr", 1)
            self.record_id_len = kwargs.get("record_id_len", 0)
            if self.block_len == v23c.DG_POST_320_BLOCK_SIZE:
                self.reserved0 = b"\0\0\0\0"

    def __getitem__(self, item: str) -> Any:
        return self.__getattribute__(item)

    def __setitem__(self, item: str, value: Any) -> None:
        self.__setattr__(item, value)

    def __bytes__(self) -> bytes:
        if self.block_len == v23c.DG_POST_320_BLOCK_SIZE:
            fmt = v23c.FMT_DATA_GROUP_POST_320
            keys = v23c.KEYS_DATA_GROUP_POST_320
        else:
            fmt = v23c.FMT_DATA_GROUP_PRE_320
            keys = v23c.KEYS_DATA_GROUP_PRE_320
        result = pack(fmt, *[self[key] for key in keys])
        return result class FileIdentificationBlock:
    """IDBLOCK class

    IDBLOCK fields

    * ``file_identification`` -  bytes : file identifier
    * ``version_str`` - bytes : format identifier
    * ``program_identification`` - bytes : creator program identifier
    * ``byte_order`` - int : integer code for byte order (endianness)
    * ``float_format`` - int : integer code for floating-point format
    * ``mdf_version`` - int : version number of MDF format
    * ``code_page`` - int : unicode code page number
    * ``reserved0`` - bytes : reserved bytes
    * ``reserved1`` - bytes : reserved bytes
    * ``unfinalized_standard_flags`` - int : standard flags for unfinalized MDF
    * ``unfinalized_custom_flags`` - int : custom flags for unfinalized MDF

    Other attributes

    * ``address`` - int : block address inside mdf file; should be 0 always

    Parameters
    ----------
    stream : file handle
        mdf file handle
    version : int
        mdf version in case of new file (dynamically created)


    """

    __slots__ = (
        "address",
        "file_identification",
        "version_str",
        "program_identification",
        "byte_order",
        "float_format",
        "mdf_version",
        "code_page",
        "reserved0",
        "reserved1",
        "unfinalized_standard_flags",
        "unfinalized_custom_flags",
    )

    def __init__(self, **kwargs) -> None:
        super().__init__()

        self.address = 0
        try:

            stream = kwargs["stream"]
            stream.seek(0)

            (
                self.file_identification,
                self.version_str,
                self.program_identification,
                self.byte_order,
                self.float_format,
                self.mdf_version,
                self.code_page,
                self.reserved0,
                self.reserved1,
                self.unfinalized_standard_flags,
                self.unfinalized_custom_flags,
            ) = unpack(v23c.ID_FMT, stream.read(v23c.ID_BLOCK_SIZE))
        except KeyError:
            version = kwargs["version"]
            self.file_identification = "MDF     ".encode("latin-1")
            self.version_str = version.encode("latin-1") + b"\0" * 4
            self.program_identification = "amdf{}".format(
                __version__.replace(".", "")
            ).encode("latin-1")
            self.byte_order = (
                v23c.BYTE_ORDER_INTEL
                if sys.byteorder == "little"
                else v23c.BYTE_ORDER_MOTOROLA
            )
            self.float_format = 0
            self.mdf_version = int(version.replace(".", ""))
            self.code_page = 0
            self.reserved0 = b"\0" * 2
            self.reserved1 = b"\0" * 26
            self.unfinalized_standard_flags = 0
            self.unfinalized_custom_flags = 0

    def __getitem__(self, item: str) -> Any:
        return self.__getattribute__(item)

    def __setitem__(self, item: str, value: Any) -> None:
        self.__setattr__(item, value)

    def __bytes__(self) -> bytes:
        result = pack(v23c.ID_FMT, *[self[key] for key in v23c.ID_KEYS])
        return result class HeaderBlock:
    """HDBLOCK class

    HDBLOCK fields

    * ``id`` - bytes : block ID; always b'HD'
    * ``block_len`` - int : block bytes size
    * ``first_dg_addr`` - int : address of first data group block (DGBLOCK)
    * ``comment_addr`` - int : address of TXBLOCK that contains the
      measurement file comment
    * ``program_addr`` - int : address of program block (PRBLOCK)
    * ``dg_nr`` - int : number of data groups
    * ``date`` - bytes : date at which the recording was started in
      "DD:MM:YYYY" format
    * ``time`` - btyes : time at which the recording was started in
      "HH:MM:SS" format
    * ``author_field`` - bytes : author name
    * ``organization_field``` - bytes : organization name
    * ``project_field``` - bytes : project name
    * ``subject_field``` - bytes : subject

    Since version 3.2 the following extra keys were added

    * ``abs_time`` - int : time stamp at which recording was started in
      nanoseconds.
    * ``tz_offset`` - int : UTC time offset in hours (= GMT time zone)
    * ``time_quality`` - int : time quality class
    * ``timer_identification`` - bytes : timer identification (time source)

    Other attributes

    * ``address`` - int : block address inside mdf file; should always be 64
    * ``comment`` - int : file comment
    * ``program`` - ProgramBlock : program block
    * ``author`` - str : measurement author
    * ``department`` - str : author department
    * ``project`` - str : working project
    * ``subject`` - str : measurement subject

    Parameters
    ----------
    stream : file handle
        mdf file handle
    version : int
        mdf version in case of new file (dynamically created)

    """

    __slots__ = (
        "address",
        "program",
        "comment",
        "id",
        "block_len",
        "first_dg_addr",
        "comment_addr",
        "program_addr",
        "dg_nr",
        "date",
        "time",
        "author",
        "department",
        "project",
        "subject",
        "abs_time",
        "tz_offset",
        "time_quality",
        "timer_identification",
        "author_field",
        "department_field",
        "project_field",
        "subject_field",
    )

    def __init__(self, **kwargs) -> None:
        super().__init__()

        self.address = 64
        self.program = None
        self.comment = ""
        try:

            stream = kwargs["stream"]
            stream.seek(64)

            (
                self.id,
                self.block_len,
                self.first_dg_addr,
                self.comment_addr,
                self.program_addr,
                self.dg_nr,
                self.date,
                self.time,
                self.author_field,
                self.department_field,
                self.project_field,
                self.subject_field,
            ) = unpack(v23c.HEADER_COMMON_FMT, stream.read(v23c.HEADER_COMMON_SIZE))

            if self.block_len > v23c.HEADER_COMMON_SIZE:
                (
                    self.abs_time,
                    self.tz_offset,
                    self.time_quality,
                    self.timer_identification,
                ) = unpack(
                    v23c.HEADER_POST_320_EXTRA_FMT,
                    stream.read(v23c.HEADER_POST_320_EXTRA_SIZE),
                )

            if self.id != b"HD":
                message = (
                    f'Expected "HD" block @{hex(self.address)} but found "{self.id}"'
                )
                message = message.format(hex(64), self.id)
                logger.exception(message)
                raise MdfException(message)

            if self.program_addr:
                self.program = ProgramBlock(address=self.program_addr, stream=stream)
            if self.comment_addr:
                self.comment = get_text_v3(address=self.comment_addr, stream=stream)

        except KeyError:
            version = kwargs.get("version", "3.20")
            self.id = b"HD"
            self.block_len = 208 if version >= "3.20" else 164
            self.first_dg_addr = 0
            self.comment_addr = 0
            self.program_addr = 0
            self.dg_nr = 0
            try:
                user = getuser()
            except ModuleNotFoundError:
                user = ""
            self.author_field = "{:\0<32}".format(user).encode("latin-1")
            self.department_field = "{:\0<32}".format("").encode("latin-1")
            self.project_field = "{:\0<32}".format("").encode("latin-1")
            self.subject_field = "{:\0<32}".format("").encode("latin-1")

            if self.block_len > v23c.HEADER_COMMON_SIZE:
                self.time_quality = 0
                self.timer_identification = "{:\0<32}".format(
                    "Local PC Reference Time"
                ).encode("latin-1")

            self.start_time = datetime(1980, 1, 1)

        self.author = self.author_field.strip(b" \r\n\t\0").decode("latin-1")
        self.department = self.department_field.strip(b" \r\n\t\0").decode("latin-1")
        self.project = self.project_field.strip(b" \r\n\t\0").decode("latin-1")
        self.subject = self.subject_field.strip(b" \r\n\t\0").decode("latin-1")

    def to_blocks(
        self,
        address: int,
        blocks: list[Any],
        defined_texts: dict[str, int],
        si_map: dict[bytes, int],
    ) -> int:
        blocks.append(self)
        self.address = address
        address += self.block_len

        key = "comment_addr"
        text = self.comment
        if text:
            if text in defined_texts:
                self[key] = defined_texts[text]
            else:
                tx_block = TextBlock(text=text)
                self[key] = address
                defined_texts[text] = address
                tx_block.address = address
                address += tx_block.block_len
                blocks.append(tx_block)
        else:
            self[key] = 0

        key = "program_addr"
        if self.program:
            self[key] = address
            address += self.program.block_len
            blocks.append(self.program)

        else:
            self[key] = 0

        self.author_field = self.author.encode("latin-1")
        self.department_field = self.department.encode("latin-1")
        self.project_field = self.project.encode("latin-1")
        self.subject_field = self.subject.encode("latin-1")

        return address

    @property
    def start_time(self) -> datetime:
        """getter and setter the measurement start timestamp

        Returns
        -------
        timestamp : datetime.datetime
            start timestamp

        """

        if self.block_len > v23c.HEADER_COMMON_SIZE:
            if self.abs_time:
                timestamp = self.abs_time / 10**9
                try:
                    timestamp = datetime.fromtimestamp(timestamp)
                except OSError:
                    timestamp = datetime.now()
            else:
                timestamp = "{} {}".format(
                    self.date.decode("ascii"), self.time.decode("ascii")
                )

                timestamp = datetime.strptime(timestamp, "%d:%m:%Y %H:%M:%S")

        else:
            timestamp = "{} {}".format(
                self.date.decode("ascii"), self.time.decode("ascii")
            )

            timestamp = datetime.strptime(timestamp, "%d:%m:%Y %H:%M:%S")

        return timestamp

    @start_time.setter
    def start_time(self, timestamp: datetime) -> None:
        self.date = timestamp.strftime("%d:%m:%Y").encode("ascii")
        self.time = timestamp.strftime("%H:%M:%S").encode("ascii")
        if self.block_len > v23c.HEADER_COMMON_SIZE:
            timestamp = int(timestamp.timestamp() * 10**9)
            self.abs_time = timestamp
            self.tz_offset = 0

    def __getitem__(self, item: str) -> Any:
        return self.__getattribute__(item)

    def __setitem__(self, item: str, value: Any) -> None:
        self.__setattr__(item, value)

    def __bytes__(self) -> bytes:
        fmt = v23c.HEADER_COMMON_FMT
        keys = v23c.HEADER_COMMON_KEYS
        if self.block_len > v23c.HEADER_COMMON_SIZE:
            fmt += v23c.HEADER_POST_320_EXTRA_FMT
            keys += v23c.HEADER_POST_320_EXTRA_KEYS
        result = pack(fmt, *[self[key] for key in keys])
        return result class TextBlock:
    """TXBLOCK class

    TXBLOCK fields

    * ``id`` - bytes : block ID; always b'TX'
    * ``block_len`` - int : block bytes size
    * ``text`` - bytes : text content

    Other attributes

    * ``address`` - int : block address inside mdf file

    Parameters
    ----------
    stream : file handle
        mdf file handle
    address : int
        block address inside mdf file
    text : bytes | str
        bytes or str for creating a new TextBlock


    Examples
    --------
    >>> tx1 = TextBlock(text='VehicleSpeed')
    >>> tx1.text_str
    'VehicleSpeed'
    >>> tx1['text']
    b'VehicleSpeed'

    """

    __slots__ = ("address", "id", "block_len", "text")

    def __init__(self, **kwargs) -> None:
        super().__init__()
        try:
            stream = kwargs["stream"]
            mapped = kwargs.get("mapped", False)
            self.address = address = kwargs["address"]
            if mapped:
                (self.id, self.block_len) = COMMON_uf(stream, address)
                if self.id != b"TX":
                    message = (
                        f'Expected "TX" block @{hex(address)} but found "{self.id}"'
                    )
                    logger.exception(message)
                    raise MdfException(message)

                self.text = stream[address + 4 : address + self.block_len]
            else:
                stream.seek(address)
                (self.id, self.block_len) = COMMON_u(stream.read(4))
                if self.id != b"TX":
                    message = (
                        f'Expected "TX" block @{hex(address)} but found "{self.id}"'
                    )
                    logger.exception(message)
                    raise MdfException(message)

                size = self.block_len - 4
                self.text = stream.read(size)

        except KeyError:
            self.address = 0
            text = kwargs["text"]

            try:
                text = text.encode("latin-1", "backslashreplace")
            except AttributeError:
                pass

            self.id = b"TX"
            self.block_len = len(text) + 5
            self.text = text + b"\0"

            if self.block_len > 65000:
                self.block_len = 65000 + 5
                self.text = self.text[:65000] + b"\0"

    def __getitem__(self, item: str) -> Any:
        return self.__getattribute__(item)

    def __setitem__(self, item: str, value: Any) -> None:
        self.__setattr__(item, value)

    def __bytes__(self) -> bytes:
        return v23c.COMMON_p(self.id, self.block_len) + self.text

    def __repr__(self) -> str:
        return (
            f"TextBlock(id={self.id},"
            f"block_len={self.block_len}, "
            f"text={self.text})"
        ) class TriggerBlock:
    """TRBLOCK class

    TRBLOCK fields

    * ``id`` - bytes : block ID; always b'TR'
    * ``block_len`` - int : block bytes size
    * ``text_addr`` - int : address of TXBLOCK that contains the trigger
      comment text
    * ``trigger_events_nr`` - int : number of trigger events
    * ``trigger_<N>_time`` - float : trigger time [s] of trigger's N-th event
    * ``trigger_<N>_pretime`` - float : pre trigger time [s] of trigger's N-th
      event
    * ``trigger_<N>_posttime`` - float : post trigger time [s] of trigger's
      N-th event

    Other attributes

    * ``address`` - int : block address inside mdf file
    * ``comment`` - str : trigger comment

    Parameters
    ----------
    stream : file handle
        mdf file handle
    address : int
        block address inside mdf file


    """

    def __init__(self, **kwargs) -> None:
        super().__init__()

        self.comment = ""

        try:
            self.address = address = kwargs["address"]
            stream = kwargs["stream"]

            stream.seek(address + 2)
            (size,) = UINT16_u(stream.read(2))
            stream.seek(address)
            block = stream.read(size)

            (self.id, self.block_len, self.text_addr, self.trigger_events_nr) = unpack(
                "<2sHIH", block[:10]
            )

            nr = self.trigger_events_nr
            if nr:
                values = unpack("<{}d".format(3 * nr), block[10:])
            for i in range(nr):
                (
                    self["trigger_{}_time".format(i)],
                    self["trigger_{}_pretime".format(i)],
                    self["trigger_{}_posttime".format(i)],
                ) = (values[i * 3], values[3 * i + 1], values[3 * i + 2])

            if self.text_addr:
                self.comment = get_text_v3(address=self.text_addr, stream=stream)

            if self.id != b"TR":
                message = f'Expected "TR" block @{hex(address)} but found "{self.id}"'

                logger.exception(message)
                raise MdfException(message)

        except KeyError:
            self.address = 0
            nr = 0
            while "trigger_{}_time".format(nr) in kwargs:
                nr += 1

            self.id = b"TR"
            self.block_len = 10 + nr * 8 * 3
            self.text_addr = 0
            self.trigger_events_nr = nr

            for i in range(nr):
                key = "trigger_{}_time".format(i)
                self[key] = kwargs[key]
                key = "trigger_{}_pretime".format(i)
                self[key] = kwargs[key]
                key = "trigger_{}_posttime".format(i)
                self[key] = kwargs[key]

    def to_blocks(self, address: int, blocks: list[Any]) -> int:
        key = "text_addr"
        text = self.comment
        if text:
            tx_block = TextBlock(text=text)
            self[key] = address
            address += tx_block.block_len
            blocks.append(tx_block)
        else:
            self[key] = 0

        blocks.append(self)
        self.address = address
        address += self.block_len

        return address

    def __getitem__(self, item: str) -> Any:
        return self.__getattribute__(item)

    def __setitem__(self, item: str, value: Any) -> None:
        self.__setattr__(item, value)

    def __bytes__(self) -> bytes:
        triggers_nr = self.trigger_events_nr
        fmt = "<2sHIH{}d".format(triggers_nr * 3)
        keys = ("id", "block_len", "text_addr", "trigger_events_nr")
        for i in range(triggers_nr):
            keys += (
                "trigger_{}_time".format(i),
                "trigger_{}_pretime".format(i),
                "trigger_{}_posttime".format(i),
            )
        result = pack(fmt, *[self[key] for key in keys])
        return result
from collections import defaultdict
from collections.abc import Iterator, Sequence
from copy import deepcopy
from io import BufferedReader, BytesIO
from itertools import product
from math import ceil
from pathlib import Path
from tempfile import TemporaryFile
from typing import Any, overload
from numpy import (
    arange,
    array,
    array_equal,
    column_stack,
    concatenate,
    dtype,
    float32,
    float64,
    frombuffer,
    linspace,
    searchsorted,
    uint16,
    unique,
    zeros,
)
from numpy.core.defchararray import decode, encode
from numpy.core.records import fromarrays, fromstring
from numpy.typing import NDArray
from pandas import DataFrame
from typing_extensions import Literal, TypedDict
from . import v2_v3_constants as v23c
from ..signal import Signal
from ..types import ChannelsType, CompressionType, RasterType, StrPathType
from ..version import __version__
from .conversion_utils import conversion_transfer
from .cutils import get_channel_raw_bytes
from .mdf_common import MDF_Common
from .options import get_global_option
from .source_utils import Source
from .utils import (
    as_non_byte_sized_signed_int,
    CHANNEL_COUNT,
    ChannelsDB,
    CONVERT,
    count_channel_groups,
    DataBlockInfo,
    fmt_to_datatype_v3,
    get_fmt_v3,
    get_text_v3,
    Group,
    is_file_like,
    MdfException,
    UniqueDB,
    validate_version_argument,
    VirtualChannelGroup,
)
from .v2_v3_blocks import (
    Channel,
    ChannelConversion,
    ChannelDependency,
    ChannelExtension,
    ChannelGroup,
    DataGroup,
    FileIdentificationBlock,
    HeaderBlock,
    TextBlock,
    TriggerBlock,
)
import logging
import mmap
import os
import sys
import time
import xml.etree.ElementTree as ET
                    new_record.append(None)

                    channel = Channel(**kargs)
                    channel.name = name
                    channel.comment = signal.comment
                    channel.source = source
                    gp_channels.append(channel)

                    self.channels_db.add(name, (dg_cntr, ch_cntr))

                    ch_cntr += 1

                    for i, (name, samples) in enumerate(
                        zip(component_names, component_samples)
                    ):

                        if i < sd_nr:
                            dep_pair = dg_cntr, ch_cntr
                            parent_dep.referenced_channels.append(dep_pair)
                            description = b"\0"
                        else:
                            description = f"{signal.name} - axis {name}"
                            description = description.encode("latin-1")

                        s_type, s_size = fmt_to_datatype_v3(samples.dtype, ())
                        shape = samples.shape[1:]

                        # source for channel
                        if signal.source:
                            source = signal.source
                            if source.source_type != 2:
                                kargs = {
                                    "type": v23c.SOURCE_ECU,
                                    "description": source.name.encode("latin-1"),
                                    "ECU_identification": source.path.encode("latin-1"),
                                }
                            else:
                                kargs = {
                                    "type": v23c.SOURCE_VECTOR,
                                    "message_name": source.name.encode("latin-1"),
                                    "sender_name": source.path.encode("latin-1"),
                                }

                            source = ChannelExtension(**kargs)

                        else:
                            source = ce_block

                        # compute additional byte offset for large records size
                        if offset > v23c.MAX_UINT16:
                            additional_byte_offset = ceil(
                                (offset - v23c.MAX_UINT16) / 8
                            )
                            start_bit_offset = offset - additional_byte_offset * 8
                        else:
                            start_bit_offset = offset
                            additional_byte_offset = 0

                        kargs = {
                            "channel_type": v23c.CHANNEL_TYPE_VALUE,
                            "data_type": s_type,
                            "start_offset": start_bit_offset,
                            "bit_count": s_size,
                            "additional_byte_offset": additional_byte_offset,
                            "block_len": channel_size,
                            "description": description,
                            "version": version,
                        }

                        if s_size < 8:
                            s_size = 8

                        new_record.append(
                            (
                                samples.dtype,
                                samples.dtype.itemsize,
                                offset // 8,
                                0,
                            )
                        )

                        channel = Channel(**kargs)
                        channel.name = name
                        channel.source = source
                        gp_channels.append(channel)

                        size = s_size
                        for dim in shape:
                            size *= dim
                        offset += size

                        self.channels_db.add(name, (dg_cntr, ch_cntr))

                        field_name = field_names.get_unique_name(name)

                        fields.append(samples)
                        types.append((field_name, samples.dtype, shape))

                        gp_dep.append(None)

                        ch_cntr += 1

        # channel group
        kargs = {
            "cycles_nr": cycles_nr,
            "samples_byte_nr": offset // 8,
            "ch_nr": ch_cntr,
        }
        if self.version >= "3.30":
            kargs["block_len"] = v23c.CG_POST_330_BLOCK_SIZE
        else:
            kargs["block_len"] = v23c.CG_PRE_330_BLOCK_SIZE
        gp.channel_group = ChannelGroup(**kargs)
        gp.channel_group.comment = comment

        # data group
        if self.version >= "3.20":
            block_len = v23c.DG_POST_320_BLOCK_SIZE
        else:
            block_len = v23c.DG_PRE_320_BLOCK_SIZE

        gp.data_group = DataGroup(block_len=block_len)